Lazy
Lazy
# Lazy class
class lazy_property(object):
""" Decorator for a lazy property of an object, i.e., an object attribute
that is determined by the result of a method call evaluated once. To
reevaluate the property, simply delete the attribute on the object, and
get it again.
"""
def __init__(self, fget):
assert not fget.__name__.startswith('__'),\
"lazy_property does not support mangled names"
self.fget = fget
def __get__(self, obj, cls):
if obj is None:
return self
value = self.fget(obj)
setattr(obj, self.fget.__name__, value)
return value
@property
def __doc__(self):
return self.fget.__doc__
@staticmethod
def reset_all(obj):
""" Reset all lazy properties on the instance `obj`. """
cls = type(obj)
obj_dict = vars(obj)
for name in list(obj_dict):
if isinstance(getattr(cls, name, None), lazy_property):
obj_dict.pop(name)
# Usage: in class Environment
class Environment:
@lazy_property
def user(self):
"""Return the current user (as an instance).
:returns: current user - sudoed
:rtype: :class:`res.users record<~odoo.addons.base.models.res_users.Users>`"""
return self(su=True)['res.users'].browse(self.uid)
Inline to get or value
Inline to get or value
uid_origin = uid_origin or (uid if isinstance(uid, int) else None)
Share class members to manage all of instances
Share class members to manage all of instances
class Registry(Mapping):
""" Model registry for a particular database.
The registry is essentially a mapping between model names and model classes.
There is one registry instance per database.
"""
_lock = threading.RLock()
_saved_lock = None
@lazy_classproperty
def registries(cls):
""" A mapping from database names to registries. """
size = config.get('registry_lru_size', None)
if not size:
# Size the LRU depending of the memory limits
if os.name != 'posix':
# cannot specify the memory limit soft on windows...
size = 42
else:
# A registry takes 10MB of memory on average, so we reserve
# 10Mb (registry) + 5Mb (working memory) per registry
avgsz = 15 * 1024 * 1024
size = int(config['limit_memory_soft'] / avgsz)
return LRU(size)
def __new__(cls, db_name):
""" Return the registry for the given database name."""
with cls._lock:
try:
return cls.registries[db_name]
except KeyError:
return cls.new(db_name)
finally:
# set db tracker - cleaned up at the WSGI dispatching phase in
# odoo.http.root
threading.current_thread().dbname = db_name
Init new instance dynamically
Init new instance dynamically
registry = object.__new__(cls)
Context manager
Context manager
class closing(AbstractContextManager):
"""Context to automatically close something at the end of a block.
Code like this:
with closing(<module>.open(<arguments>)) as f:
<block>
is equivalent to this:
f = <module>.open(<arguments>)
try:
<block>
finally:
f.close()
"""
def __init__(self, thing):
self.thing = thing
def __enter__(self):
return self.thing
def __exit__(self, *exc_info):
self.thing.close()
- Context with
yield
,finally
:
@contextmanager
def protecting(self, what, records=None):
""" Prevent the invalidation or recomputation of fields on records.
The parameters are either:
- ``what`` a collection of fields and ``records`` a recordset, or
- ``what`` a collection of pairs ``(fields, records)``.
"""
protected = self._protected
try:
protected.pushmap()
if records is not None: # Handle first signature
ids_by_field = {field: records._ids for field in what}
else: # Handle second signature
ids_by_field = defaultdict(list)
for fields, what_records in what:
for field in fields:
ids_by_field[field].extend(what_records._ids)
for field, rec_ids in ids_by_field.items():
ids = protected.get(field)
protected[field] = ids.union(rec_ids) if ids else frozenset(rec_ids)
yield
finally:
protected.popmap()
Decorator
Decorator
def attrsetter(attr, value):
""" Return a function that sets ``attr`` on its argument and returns it. """
return lambda method: setattr(method, attr, value) or method
def constrains(*args):
"""Decorate a constraint checker.
Each argument must be a field name used in the check::
@api.constrains('name', 'description')
def _check_description(self):
for record in self:
if record.name == record.description:
raise ValidationError("Fields name and description must be different")
Invoked on the records on which one of the named fields has been modified.
Should raise :exc:`~odoo.exceptions.ValidationError` if the
validation failed.
.. warning::
``@constrains`` only supports simple field names, dotted names
(fields of relational fields e.g. ``partner_id.customer``) are not
supported and will be ignored.
``@constrains`` will be triggered only if the declared fields in the
decorated method are included in the ``create`` or ``write`` call.
It implies that fields not present in a view will not trigger a call
during a record creation. A override of ``create`` is necessary to make
sure a constraint will always be triggered (e.g. to test the absence of
value).
One may also pass a single function as argument. In that case, the field
names are given by calling the function with a model instance.
"""
if args and callable(args[0]):
args = args[0]
return attrsetter('_constrains', args)
@api.constrains('code')
def _check_account_code(self):
for account in self:
if not re.match(ACCOUNT_CODE_REGEX, account.code):
raise ValidationError(_(
"The account code can only contain alphanumeric characters and dots."
))
@api.constrains('account_type')
def _check_account_is_bank_journal_bank_account(self):
self.env['account.account'].flush_model(['account_type'])
self.env['account.journal'].flush_model(['type', 'default_account_id'])
Attact data to a thread
Attact data to a thread
class SQLCollector(Collector):
"""
Saves all executed queries in the current thread with the call stack.
"""
name = 'sql'
def start(self):
init_thread = self.profiler.init_thread
if not hasattr(init_thread, 'query_hooks'):
init_thread.query_hooks = []
init_thread.query_hooks.append(self.hook)
class Cursor(BaseCursor):
def execute(self, query, params=None, log_exceptions=True):
...
current_thread = threading.current_thread()
if hasattr(current_thread, 'query_count'):
current_thread.query_count += 1
current_thread.query_time += delay
# optional hooks for performance and tracing analysis
for hook in getattr(current_thread, 'query_hooks', ()):
hook(self, query, params, start, delay)
Hold/Share data at class level
Hold/Share data at class level
class Registry(Mapping):
""" Model registry for a particular database.
The registry is essentially a mapping between model names and model classes.
There is one registry instance per database.
"""
_lock = threading.RLock()
_saved_lock = None
@lazy_classproperty
def registries(cls):
""" A mapping from database names to registries. """
size = config.get('registry_lru_size', None)
if not size:
# Size the LRU depending of the memory limits
if os.name != 'posix':
# cannot specify the memory limit soft on windows...
size = 42
else:
# A registry takes 10MB of memory on average, so we reserve
# 10Mb (registry) + 5Mb (working memory) per registry
avgsz = 15 * 1024 * 1024
size = int(config['limit_memory_soft'] / avgsz)
return LRU(size)
def __new__(cls, db_name):
""" Return the registry for the given database name."""
with cls._lock:
try:
return cls.registries[db_name]
except KeyError:
return cls.new(db_name)
finally:
# set db tracker - cleaned up at the WSGI dispatching phase in
# odoo.http.root
threading.current_thread().dbname = db_name
- Be carefull when we change pointer of object
class ClassSample:
shared_counter = 0
def __new__(cls, *args, **kwargs):
cls.shared_counter += 1
return object.__new__(cls)
def __init__(self):
pass
def count(self): # OK
type(self).shared_counter += 1
class_instance = ClassSample()
# class_instance.shared_counter += 1 # BECAREFULL: shared_counter of class instance will be copied from `cls`
class_instance.count() # OK
print(ClassSample.shared_counter, class_instance.shared_counter)
Loading module dynamically (in the system)
Loading module dynamically (in the system)
qualname = f'odoo.addons.{module_name}'
if qualname in sys.modules:
return
try:
__import__(qualname)
# Show all of imported modules
import sys
sys.modules.keys()
Metaclass
Metaclass
- Used for hook (value)
- User for control class instance
class MetaclassToCreateClass(type):
we_want_to_manage_class_instances = []
def __new__(meta, name, bases, attrs):
# Hook in creation class (NOT class instance)
# - meta: instance of type (isinstance(meta, type) >> True)
# - name: ClassSample (the class's name which indicate metaclass)
# - attrs - bags of attributes (methods, attributes...) of the class
# - return: MetaclassToCreateClass type
# The logic is creation new instance of type class (named `meta`) then update
# all attributes (from attrs, name) and bases of this type instance
attrs['_hai_hooked'] = "this is sample hook"
meta_instance = type.__new__(meta, name, bases, attrs)
return meta_instance
def __init__(self, name, bases, attrs):
# `self` is MetaclassToCreateClass instance (isinstance(self, MetaclassToCreateClass) >> True)
# May be change `self` if need
# Hook to manage class instance
self.we_want_to_manage_class_instances.append(self)
# `self` is `meta_instance` which created from `__new__`
super().__init__(name, bases, attrs)
class ClassSample(metaclass=MetaclassToCreateClass):
def __init__(self):
pass
class ClassSample2(metaclass=MetaclassToCreateClass):
def __init__(self):
pass
class_instance = ClassSample()
print(ClassSample._hai_hooked)
print(class_instance._hai_hooked)
assert id(ClassSample._hai_hooked) == id(class_instance._hai_hooked) # same
print(MetaclassToCreateClass.we_want_to_manage_class_instances) # ClassSample, ClassSample2
- We can define class instance directly (dynamically)
ModelClass = type(name, (tuple_inherited_bases, ), {
'_name': name,
'_register': False,
'_original_module': cls._module,
'_inherit_module': {}, # map parent to introducing module
'_inherit_children': OrderedSet(), # names of children models
'_inherits_children': set(), # names of children models
'_fields': {}, # populated in _setup_base()
})
Using *args
Using *args
# From caller
return self.descendants(model_names, '_inherit', '_inherits') # *kinds is enumeration
def descendants(self, model_names, *kinds):
""" Return the models corresponding to ``model_names`` and all those
that inherit/inherits from them.
"""
assert all(kind in ('_inherit', '_inherits') for kind in kinds) # use `kinds`, NOT `*kinds`
funcs = [attrgetter(kind + '_children') for kind in kinds]
Use `queue`
Use queue
queue = deque(model_names)
while queue:
model = self[queue.popleft()]
models.add(model._name)
for func in funcs:
queue.extend(func(model))
return models
Using `property`
Using property
@property
def ids(self):
""" Return the list of actual record ids corresponding to ``self``. """
return list(origin_ids(self._ids))
# backward-compatibility with former browse records
_cr = property(lambda self: self.env.cr) # fget = lambda self: self.env.cr
yield` usage
yield
usage
- Create
enumerator
object to hold a pointer to current record - When reading data from the
enumerator
then the pointer will be move the next one. When get the end, theenumerator
will return empty array (read only ONE
)
def generate(data):
index = 0
while index < len(data):
yield {
'data': data[index]
}
index += 1
Lock
Lock
_lock = threading.RLock()
_saved_lock = None
@lazy_classproperty
def registries(cls):
""" A mapping from database names to registries. """
size = config.get('registry_lru_size', None)
if not size:
# Size the LRU depending of the memory limits
if os.name != 'posix':
# cannot specify the memory limit soft on windows...
size = 42
else:
# A registry takes 10MB of memory on average, so we reserve
# 10Mb (registry) + 5Mb (working memory) per registry
avgsz = 15 * 1024 * 1024
size = int(config['limit_memory_soft'] / avgsz)
return LRU(size)
def __new__(cls, db_name):
""" Return the registry for the given database name."""
with cls._lock:
try:
return cls.registries[db_name]
except KeyError:
return cls.new(db_name)
finally:
# set db tracker - cleaned up at the WSGI dispatching phase in
# odoo.http.root
threading.current_thread().dbname = db_name
Build partial (partly or postpend call)
Build partial (partly or postpend call)
- The `partial` has full or half of parameters. But we will call later (NOT NOW)
class Registry:
...
def post_init(self, func, *args, **kwargs):
""" Register a function to call at the end of :meth:`~.init_models`. """
self._post_init_queue.append(partial(func, *args, **kwargs))
...
def init_models(self):
while self._post_init_queue:
func = self._post_init_queue.popleft()
func() # CALL NOW
SQL insert/update by batch
SQL insert/update by batch
INSERT INTO ir_model_data (module,name,model,res_id,noupdate)
VALUES (%s,%s,%s,%s,%s), (%s,%s,%s,%s,%s), (%s,%s,%s,%s,%s), (%s,%s,%s,%s,%s), (%s,%s,%s,%s,%s)
ON CONFLICT (module, name)
DO UPDATE SET (model, res_id, write_date) =
(EXCLUDED.model, EXCLUDED.res_id, now() at time zone 'UTC')
WHERE (ir_model_data.res_id != EXCLUDED.res_id OR ir_model_data.model != EXCLUDED.model)
RETURNING module, name, model, res_id, create_date, write_date
self.env.cr.execute(query, [arg for row in sub_rows for arg in row])
Creation of object
Creation of object
class Stream:
"""
Send the content of a file, an attachment or a binary field via HTTP
This utility is safe, cache-aware and uses the best available
streaming strategy. Works best with the --x-sendfile cli option.
Create a Stream via one of the constructors: :meth:`~from_path`:,
:meth:`~from_attachment`: or :meth:`~from_binary_field`:, generate
the corresponding HTTP response object via :meth:`~get_response`:.
Instantiating a Stream object manually without using one of the
dedicated constructors is discouraged.
"""
type: str = '' # 'data' or 'path' or 'url'
data = None
path = None
url = None
mimetype = None
as_attachment = False
download_name = None
conditional = True
etag = True
last_modified = None
max_age = None
immutable = False
size = None
public = False
def __init__(self, **kwargs):
self.__dict__.update(kwargs)
Other (temp notes)
Other (temp notes)
Server [localhost]: localhost
Database [postgres]:
Port [5432]:
Username [postgres]:
Password for user postgres: hai
psql (16.4)
WARNING: Console code page (437) differs from Windows code page (1252)
8-bit characters might not work correctly. See psql reference
page "Notes for Windows users" for details.
Type "help" for help.
postgres=#