Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py # ext/__init__.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php from .. import util as _sa_util _sa_util.dependencies.resolve_all("sqlalchemy.ext")	322 B	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 associationproxy.py # ext/associationproxy.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Contain the ``AssociationProxy`` class. The ``AssociationProxy`` is a Python property object which provides transparent proxied access to the endpoint of an association object. See the example ``examples/association/proxied_association.py``. """ import operator from .. import exc from .. import inspect from .. import orm from .. import util from ..orm import collections from ..orm import interfaces from ..sql import or_ from ..sql.operators import ColumnOperators def association_proxy(target_collection, attr, *kw): r"""Return a Python property implementing a view of a target attribute which references an attribute on members of the target. The returned value is an instance of :class:`.AssociationProxy`. Implements a Python property representing a relationship as a collection of simpler values, or a scalar value. The proxied property will mimic the collection type of the target (list, dict or set), or, in the case of a one to one relationship, a simple scalar value. :param target_collection: Name of the attribute we'll proxy to. This attribute is typically mapped by :func:`~sqlalchemy.orm.relationship` to link to a target collection, but can also be a many-to-one or non-scalar relationship. :param attr: Attribute on the associated instance or instances we'll proxy for. For example, given a target collection of [obj1, obj2], a list created by this proxy property would look like [getattr(obj1, attr), getattr(obj2, attr)] If the relationship is one-to-one or otherwise uselist=False, then simply: getattr(obj, attr) :param creator: optional. When new items are added to this proxied collection, new instances of the class collected by the target collection will be created. For list and set collections, the target class constructor will be called with the 'value' for the new instance. For dict types, two arguments are passed: key and value. If you want to construct instances differently, supply a creator* function that takes arguments as above and returns instances. For scalar relationships, creator() will be called if the target is None. If the target is present, set operations are proxied to setattr() on the associated object. If you have an associated object with multiple attributes, you may set up multiple association proxies mapping to different attributes. See the unit tests for examples, and for examples of how creator() functions can be used to construct the scalar relationship on-demand in this situation. :param \\kw: Passes along any other keyword arguments to :class:`.AssociationProxy`. """ return AssociationProxy(target_collection, attr, *kw) ASSOCIATION_PROXY = util.symbol("ASSOCIATION_PROXY") """Symbol indicating an :class:`.InspectionAttr` that's of type :class:`.AssociationProxy`. Is assigned to the :attr:`.InspectionAttr.extension_type` attribute. """ class AssociationProxy(interfaces.InspectionAttrInfo): """A descriptor that presents a read/write view of an object attribute.""" is_attribute = True extension_type = ASSOCIATION_PROXY def __init__( self, target_collection, attr, creator=None, getset_factory=None, proxy_factory=None, proxy_bulk_set=None, info=None, cascade_scalar_deletes=False, ): """Construct a new :class:`.AssociationProxy`. The :func:`.association_proxy` function is provided as the usual entrypoint here, though :class:`.AssociationProxy` can be instantiated and/or subclassed directly. :param target_collection: Name of the collection we'll proxy to, usually created with :func:`_orm.relationship`. :param attr: Attribute on the collected instances we'll proxy for. For example, given a target collection of [obj1, obj2], a list created by this proxy property would look like [getattr(obj1, attr), getattr(obj2, attr)] :param creator: Optional. When new items are added to this proxied collection, new instances of the class collected by the target collection will be created. For list and set collections, the target class constructor will be called with the 'value' for the new instance. For dict types, two arguments are passed: key and value. If you want to construct instances differently, supply a 'creator' function that takes arguments as above and returns instances. :param cascade_scalar_deletes: when True, indicates that setting the proxied value to ``None``, or deleting it via ``del``, should also remove the source object. Only applies to scalar attributes. Normally, removing the proxied target will not remove the proxy source, as this object may have other state that is still to be kept. .. versionadded:: 1.3 .. seealso:: :ref:`cascade_scalar_deletes` - complete usage example :param getset_factory: Optional. Proxied attribute access is automatically handled by routines that get and set values based on the `attr` argument for this proxy. If you would like to customize this behavior, you may supply a `getset_factory` callable that produces a tuple of `getter` and `setter` functions. The factory is called with two arguments, the abstract type of the underlying collection and this proxy instance. :param proxy_factory: Optional. The type of collection to emulate is determined by sniffing the target collection. If your collection type can't be determined by duck typing or you'd like to use a different collection implementation, you may supply a factory function to produce those collections. Only applicable to non-scalar relationships. :param proxy_bulk_set: Optional, use with proxy_factory. See the _set() method for details. :param info: optional, will be assigned to :attr:`.AssociationProxy.info` if present. .. versionadded:: 1.0.9 """ self.target_collection = target_collection self.value_attr = attr self.creator = creator self.getset_factory = getset_factory self.proxy_factory = proxy_factory self.proxy_bulk_set = proxy_bulk_set self.cascade_scalar_deletes = cascade_scalar_deletes self.key = "_%s_%s_%s" % ( type(self).__name__, target_collection, id(self), ) if info: self.info = info def __get__(self, obj, class_): if class_ is None: return self inst = self._as_instance(class_, obj) if inst: return inst.get(obj) # obj has to be None here # assert obj is None return self def __set__(self, obj, values): class_ = type(obj) return self._as_instance(class_, obj).set(obj, values) def __delete__(self, obj): class_ = type(obj) return self._as_instance(class_, obj).delete(obj) def for_class(self, class_, obj=None): r"""Return the internal state local to a specific mapped class. E.g., given a class ``User``:: class User(Base): # ... keywords = association_proxy('kws', 'keyword') If we access this :class:`.AssociationProxy` from :attr:`_orm.Mapper.all_orm_descriptors`, and we want to view the target class for this proxy as mapped by ``User``:: inspect(User).all_orm_descriptors["keywords"].for_class(User).target_class This returns an instance of :class:`.AssociationProxyInstance` that is specific to the ``User`` class. The :class:`.AssociationProxy` object remains agnostic of its parent class. :param class\_: the class that we are returning state for. :param obj: optional, an instance of the class that is required if the attribute refers to a polymorphic target, e.g. where we have to look at the type of the actual destination object to get the complete path. .. versionadded:: 1.3 - :class:`.AssociationProxy` no longer stores any state specific to a particular parent class; the state is now stored in per-class :class:`.AssociationProxyInstance` objects. """ return self._as_instance(class_, obj) def _as_instance(self, class_, obj): try: inst = class_.__dict__[self.key + "_inst"] except KeyError: inst = None # avoid exception context if inst is None: owner = self._calc_owner(class_) if owner is not None: inst = AssociationProxyInstance.for_proxy(self, owner, obj) setattr(class_, self.key + "_inst", inst) else: inst = None if inst is not None and not inst._is_canonical: # the AssociationProxyInstance can't be generalized # since the proxied attribute is not on the targeted # class, only on subclasses of it, which might be # different. only return for the specific # object's current value return inst._non_canonical_get_for_object(obj) else: return inst def _calc_owner(self, target_cls): # we might be getting invoked for a subclass # that is not mapped yet, in some declarative situations. # save until we are mapped try: insp = inspect(target_cls) except exc.NoInspectionAvailable: # can't find a mapper, don't set owner. if we are a not-yet-mapped # subclass, we can also scan through __mro__ to find a mapped # class, but instead just wait for us to be called again against a # mapped class normally. return None else: return insp.mapper.class_manager.class_ def _default_getset(self, collection_class): attr = self.value_attr _getter = operator.attrgetter(attr) def getter(target): return _getter(target) if target is not None else None if collection_class is dict: def setter(o, k, v): setattr(o, attr, v) else: def setter(o, v): setattr(o, attr, v) return getter, setter def __repr__(self): return "AssociationProxy(%r, %r)" % ( self.target_collection, self.value_attr, ) class AssociationProxyInstance(object): """A per-class object that serves class- and object-specific results. This is used by :class:`.AssociationProxy` when it is invoked in terms of a specific class or instance of a class, i.e. when it is used as a regular Python descriptor. When referring to the :class:`.AssociationProxy` as a normal Python descriptor, the :class:`.AssociationProxyInstance` is the object that actually serves the information. Under normal circumstances, its presence is transparent:: >>> User.keywords.scalar False In the special case that the :class:`.AssociationProxy` object is being accessed directly, in order to get an explicit handle to the :class:`.AssociationProxyInstance`, use the :meth:`.AssociationProxy.for_class` method:: proxy_state = inspect(User).all_orm_descriptors["keywords"].for_class(User) # view if proxy object is scalar or not >>> proxy_state.scalar False .. versionadded:: 1.3 """ # noqa def __init__(self, parent, owning_class, target_class, value_attr): self.parent = parent self.key = parent.key self.owning_class = owning_class self.target_collection = parent.target_collection self.collection_class = None self.target_class = target_class self.value_attr = value_attr target_class = None """The intermediary class handled by this :class:`.AssociationProxyInstance`. Intercepted append/set/assignment events will result in the generation of new instances of this class. """ @classmethod def for_proxy(cls, parent, owning_class, parent_instance): target_collection = parent.target_collection value_attr = parent.value_attr prop = orm.class_mapper(owning_class).get_property(target_collection) # this was never asserted before but this should be made clear. if not isinstance(prop, orm.RelationshipProperty): util.raise_( NotImplementedError( "association proxy to a non-relationship " "intermediary is not supported" ), replace_context=None, ) target_class = prop.mapper.class_ try: target_assoc = cls._cls_unwrap_target_assoc_proxy( target_class, value_attr ) except AttributeError: # the proxied attribute doesn't exist on the target class; # return an "ambiguous" instance that will work on a per-object # basis return AmbiguousAssociationProxyInstance( parent, owning_class, target_class, value_attr ) else: return cls._construct_for_assoc( target_assoc, parent, owning_class, target_class, value_attr ) @classmethod def _construct_for_assoc( cls, target_assoc, parent, owning_class, target_class, value_attr ): if target_assoc is not None: return ObjectAssociationProxyInstance( parent, owning_class, target_class, value_attr ) attr = getattr(target_class, value_attr) if not hasattr(attr, "_is_internal_proxy"): return AmbiguousAssociationProxyInstance( parent, owning_class, target_class, value_attr ) is_object = attr._impl_uses_objects if is_object: return ObjectAssociationProxyInstance( parent, owning_class, target_class, value_attr ) else: return ColumnAssociationProxyInstance( parent, owning_class, target_class, value_attr ) def _get_property(self): return orm.class_mapper(self.owning_class).get_property( self.target_collection ) @property def _comparator(self): return self._get_property().comparator def __clause_element__(self): raise NotImplementedError( "The association proxy can't be used as a plain column " "expression; it only works inside of a comparison expression" ) @classmethod def _cls_unwrap_target_assoc_proxy(cls, target_class, value_attr): attr = getattr(target_class, value_attr) if isinstance(attr, (AssociationProxy, AssociationProxyInstance)): return attr return None @util.memoized_property def _unwrap_target_assoc_proxy(self): return self._cls_unwrap_target_assoc_proxy( self.target_class, self.value_attr ) @property def remote_attr(self): """The 'remote' class attribute referenced by this :class:`.AssociationProxyInstance`. .. seealso:: :attr:`.AssociationProxyInstance.attr` :attr:`.AssociationProxyInstance.local_attr` """ return getattr(self.target_class, self.value_attr) @property def local_attr(self): """The 'local' class attribute referenced by this :class:`.AssociationProxyInstance`. .. seealso:: :attr:`.AssociationProxyInstance.attr` :attr:`.AssociationProxyInstance.remote_attr` """ return getattr(self.owning_class, self.target_collection) @property def attr(self): """Return a tuple of ``(local_attr, remote_attr)``. This attribute is convenient when specifying a join using :meth:`_query.Query.join` across two relationships:: sess.query(Parent).join(Parent.proxied.attr) .. seealso:: :attr:`.AssociationProxyInstance.local_attr` :attr:`.AssociationProxyInstance.remote_attr` """ return (self.local_attr, self.remote_attr) @util.memoized_property def scalar(self): """Return ``True`` if this :class:`.AssociationProxyInstance` proxies a scalar relationship on the local side.""" scalar = not self._get_property().uselist if scalar: self._initialize_scalar_accessors() return scalar @util.memoized_property def _value_is_scalar(self): return ( not self._get_property() .mapper.get_property(self.value_attr) .uselist ) @property def _target_is_object(self): raise NotImplementedError() def _initialize_scalar_accessors(self): if self.parent.getset_factory: get, set_ = self.parent.getset_factory(None, self) else: get, set_ = self.parent._default_getset(None) self._scalar_get, self._scalar_set = get, set_ def _default_getset(self, collection_class): attr = self.value_attr _getter = operator.attrgetter(attr) def getter(target): return _getter(target) if target is not None else None if collection_class is dict: def setter(o, k, v): return setattr(o, attr, v) else: def setter(o, v): return setattr(o, attr, v) return getter, setter @property def info(self): return self.parent.info def get(self, obj): if obj is None: return self if self.scalar: target = getattr(obj, self.target_collection) return self._scalar_get(target) else: try: # If the owning instance is reborn (orm session resurrect, # etc.), refresh the proxy cache. creator_id, self_id, proxy = getattr(obj, self.key) except AttributeError: pass else: if id(obj) == creator_id and id(self) == self_id: assert self.collection_class is not None return proxy self.collection_class, proxy = self._new( _lazy_collection(obj, self.target_collection) ) setattr(obj, self.key, (id(obj), id(self), proxy)) return proxy def set(self, obj, values): if self.scalar: creator = ( self.parent.creator if self.parent.creator else self.target_class ) target = getattr(obj, self.target_collection) if target is None: if values is None: return setattr(obj, self.target_collection, creator(values)) else: self._scalar_set(target, values) if values is None and self.parent.cascade_scalar_deletes: setattr(obj, self.target_collection, None) else: proxy = self.get(obj) assert self.collection_class is not None if proxy is not values: proxy._bulk_replace(self, values) def delete(self, obj): if self.owning_class is None: self._calc_owner(obj, None) if self.scalar: target = getattr(obj, self.target_collection) if target is not None: delattr(target, self.value_attr) delattr(obj, self.target_collection) def _new(self, lazy_collection): creator = ( self.parent.creator if self.parent.creator else self.target_class ) collection_class = util.duck_type_collection(lazy_collection()) if self.parent.proxy_factory: return ( collection_class, self.parent.proxy_factory( lazy_collection, creator, self.value_attr, self ), ) if self.parent.getset_factory: getter, setter = self.parent.getset_factory(collection_class, self) else: getter, setter = self.parent._default_getset(collection_class) if collection_class is list: return ( collection_class, _AssociationList( lazy_collection, creator, getter, setter, self ), ) elif collection_class is dict: return ( collection_class, _AssociationDict( lazy_collection, creator, getter, setter, self ), ) elif collection_class is set: return ( collection_class, _AssociationSet( lazy_collection, creator, getter, setter, self ), ) else: raise exc.ArgumentError( "could not guess which interface to use for " 'collection_class "%s" backing "%s"; specify a ' "proxy_factory and proxy_bulk_set manually" % (self.collection_class.__name__, self.target_collection) ) def _set(self, proxy, values): if self.parent.proxy_bulk_set: self.parent.proxy_bulk_set(proxy, values) elif self.collection_class is list: proxy.extend(values) elif self.collection_class is dict: proxy.update(values) elif self.collection_class is set: proxy.update(values) else: raise exc.ArgumentError( "no proxy_bulk_set supplied for custom " "collection_class implementation" ) def _inflate(self, proxy): creator = ( self.parent.creator and self.parent.creator or self.target_class ) if self.parent.getset_factory: getter, setter = self.parent.getset_factory( self.collection_class, self ) else: getter, setter = self.parent._default_getset(self.collection_class) proxy.creator = creator proxy.getter = getter proxy.setter = setter def _criterion_exists(self, criterion=None, kwargs): is_has = kwargs.pop("is_has", None) target_assoc = self._unwrap_target_assoc_proxy if target_assoc is not None: inner = target_assoc._criterion_exists( criterion=criterion, kwargs ) return self._comparator._criterion_exists(inner) if self._target_is_object: prop = getattr(self.target_class, self.value_attr) value_expr = prop._criterion_exists(criterion, kwargs) else: if kwargs: raise exc.ArgumentError( "Can't apply keyword arguments to column-targeted " "association proxy; use ==" ) elif is_has and criterion is not None: raise exc.ArgumentError( "Non-empty has() not allowed for " "column-targeted association proxy; use ==" ) value_expr = criterion return self._comparator._criterion_exists(value_expr) def any(self, criterion=None, kwargs): """Produce a proxied 'any' expression using EXISTS. This expression will be a composed product using the :meth:`.RelationshipProperty.Comparator.any` and/or :meth:`.RelationshipProperty.Comparator.has` operators of the underlying proxied attributes. """ if self._unwrap_target_assoc_proxy is None and ( self.scalar and (not self._target_is_object or self._value_is_scalar) ): raise exc.InvalidRequestError( "'any()' not implemented for scalar " "attributes. Use has()." ) return self._criterion_exists( criterion=criterion, is_has=False, kwargs ) def has(self, criterion=None, kwargs): """Produce a proxied 'has' expression using EXISTS. This expression will be a composed product using the :meth:`.RelationshipProperty.Comparator.any` and/or :meth:`.RelationshipProperty.Comparator.has` operators of the underlying proxied attributes. """ if self._unwrap_target_assoc_proxy is None and ( not self.scalar or (self._target_is_object and not self._value_is_scalar) ): raise exc.InvalidRequestError( "'has()' not implemented for collections. " "Use any()." ) return self._criterion_exists( criterion=criterion, is_has=True, kwargs ) def __repr__(self): return "%s(%r)" % (self.__class__.__name__, self.parent) class AmbiguousAssociationProxyInstance(AssociationProxyInstance): """an :class:`.AssociationProxyInstance` where we cannot determine the type of target object. """ _is_canonical = False def _ambiguous(self): raise AttributeError( "Association proxy %s.%s refers to an attribute '%s' that is not " "directly mapped on class %s; therefore this operation cannot " "proceed since we don't know what type of object is referred " "towards" % ( self.owning_class.__name__, self.target_collection, self.value_attr, self.target_class, ) ) def get(self, obj): if obj is None: return self else: return super(AmbiguousAssociationProxyInstance, self).get(obj) def __eq__(self, obj): self._ambiguous() def __ne__(self, obj): self._ambiguous() def any(self, criterion=None, kwargs): self._ambiguous() def has(self, criterion=None, kwargs): self._ambiguous() @util.memoized_property def _lookup_cache(self): # mapping of <subclass>->AssociationProxyInstance. # e.g. proxy is A-> A.b -> B -> B.b_attr, but B.b_attr doesn't exist; # only B1(B) and B2(B) have "b_attr", keys in here would be B1, B2 return {} def _non_canonical_get_for_object(self, parent_instance): if parent_instance is not None: actual_obj = getattr(parent_instance, self.target_collection) if actual_obj is not None: try: insp = inspect(actual_obj) except exc.NoInspectionAvailable: pass else: mapper = insp.mapper instance_class = mapper.class_ if instance_class not in self._lookup_cache: self._populate_cache(instance_class, mapper) try: return self._lookup_cache[instance_class] except KeyError: pass # no object or ambiguous object given, so return "self", which # is a proxy with generally only instance-level functionality return self def _populate_cache(self, instance_class, mapper): prop = orm.class_mapper(self.owning_class).get_property( self.target_collection ) if mapper.isa(prop.mapper): target_class = instance_class try: target_assoc = self._cls_unwrap_target_assoc_proxy( target_class, self.value_attr ) except AttributeError: pass else: self._lookup_cache[instance_class] = self._construct_for_assoc( target_assoc, self.parent, self.owning_class, target_class, self.value_attr, ) class ObjectAssociationProxyInstance(AssociationProxyInstance): """an :class:`.AssociationProxyInstance` that has an object as a target.""" _target_is_object = True _is_canonical = True def contains(self, obj): """Produce a proxied 'contains' expression using EXISTS. This expression will be a composed product using the :meth:`.RelationshipProperty.Comparator.any`, :meth:`.RelationshipProperty.Comparator.has`, and/or :meth:`.RelationshipProperty.Comparator.contains` operators of the underlying proxied attributes. """ target_assoc = self._unwrap_target_assoc_proxy if target_assoc is not None: return self._comparator._criterion_exists( target_assoc.contains(obj) if not target_assoc.scalar else target_assoc == obj ) elif ( self._target_is_object and self.scalar and not self._value_is_scalar ): return self._comparator.has( getattr(self.target_class, self.value_attr).contains(obj) ) elif self._target_is_object and self.scalar and self._value_is_scalar: raise exc.InvalidRequestError( "contains() doesn't apply to a scalar object endpoint; use ==" ) else: return self._comparator._criterion_exists({self.value_attr: obj}) def __eq__(self, obj): # note the has() here will fail for collections; eq_() # is only allowed with a scalar. if obj is None: return or_( self._comparator.has({self.value_attr: obj}), self._comparator == None, ) else: return self._comparator.has({self.value_attr: obj}) def __ne__(self, obj): # note the has() here will fail for collections; eq_() # is only allowed with a scalar. return self._comparator.has( getattr(self.target_class, self.value_attr) != obj ) class ColumnAssociationProxyInstance( ColumnOperators, AssociationProxyInstance ): """an :class:`.AssociationProxyInstance` that has a database column as a target. """ _target_is_object = False _is_canonical = True def __eq__(self, other): # special case "is None" to check for no related row as well expr = self._criterion_exists( self.remote_attr.operate(operator.eq, other) ) if other is None: return or_(expr, self._comparator == None) else: return expr def operate(self, op, other, kwargs): return self._criterion_exists( self.remote_attr.operate(op, other, *kwargs) ) class _lazy_collection(object): def __init__(self, obj, target): self.parent = obj self.target = target def __call__(self): return getattr(self.parent, self.target) def __getstate__(self): return {"obj": self.parent, "target": self.target} def __setstate__(self, state): self.parent = state["obj"] self.target = state["target"] class _AssociationCollection(object): def __init__(self, lazy_collection, creator, getter, setter, parent): """Constructs an _AssociationCollection. This will always be a subclass of either _AssociationList, _AssociationSet, or _AssociationDict. lazy_collection A callable returning a list-based collection of entities (usually an object attribute managed by a SQLAlchemy relationship()) creator A function that creates new target entities. Given one parameter: value. This assertion is assumed:: obj = creator(somevalue) assert getter(obj) == somevalue getter A function. Given an associated object, return the 'value'. setter A function. Given an associated object and a value, store that value on the object. """ self.lazy_collection = lazy_collection self.creator = creator self.getter = getter self.setter = setter self.parent = parent col = property(lambda self: self.lazy_collection()) def __len__(self): return len(self.col) def __bool__(self): return bool(self.col) __nonzero__ = __bool__ def __getstate__(self): return {"parent": self.parent, "lazy_collection": self.lazy_collection} def __setstate__(self, state): self.parent = state["parent"] self.lazy_collection = state["lazy_collection"] self.parent._inflate(self) def _bulk_replace(self, assoc_proxy, values): self.clear() assoc_proxy._set(self, values) class _AssociationList(_AssociationCollection): """Generic, converting, list-to-list proxy.""" def _create(self, value): return self.creator(value) def _get(self, object_): return self.getter(object_) def _set(self, object_, value): return self.setter(object_, value) def __getitem__(self, index): if not isinstance(index, slice): return self._get(self.col[index]) else: return [self._get(member) for member in self.col[index]] def __setitem__(self, index, value): if not isinstance(index, slice): self._set(self.col[index], value) else: if index.stop is None: stop = len(self) elif index.stop < 0: stop = len(self) + index.stop else: stop = index.stop step = index.step or 1 start = index.start or 0 rng = list(range(index.start or 0, stop, step)) if step == 1: for i in rng: del self[start] i = start for item in value: self.insert(i, item) i += 1 else: if len(value) != len(rng): raise ValueError( "attempt to assign sequence of size %s to " "extended slice of size %s" % (len(value), len(rng)) ) for i, item in zip(rng, value): self._set(self.col[i], item) def __delitem__(self, index): del self.col[index] def __contains__(self, value): for member in self.col: # testlib.pragma exempt:__eq__ if self._get(member) == value: return True return False def __getslice__(self, start, end): return [self._get(member) for member in self.col[start:end]] def __setslice__(self, start, end, values): members = [self._create(v) for v in values] self.col[start:end] = members def __delslice__(self, start, end): del self.col[start:end] def __iter__(self): """Iterate over proxied values. For the actual domain objects, iterate over .col instead or just use the underlying collection directly from its property on the parent. """ for member in self.col: yield self._get(member) return def append(self, value): col = self.col item = self._create(value) col.append(item) def count(self, value): return sum( [ 1 for _ in util.itertools_filter( lambda v: v == value, iter(self) ) ] ) def extend(self, values): for v in values: self.append(v) def insert(self, index, value): self.col[index:index] = [self._create(value)] def pop(self, index=-1): return self.getter(self.col.pop(index)) def remove(self, value): for i, val in enumerate(self): if val == value: del self.col[i] return raise ValueError("value not in list") def reverse(self): """Not supported, use reversed(mylist)""" raise NotImplementedError def sort(self): """Not supported, use sorted(mylist)""" raise NotImplementedError def clear(self): del self.col[0 : len(self.col)] def __eq__(self, other): return list(self) == other def __ne__(self, other): return list(self) != other def __lt__(self, other): return list(self) < other def __le__(self, other): return list(self) <= other def __gt__(self, other): return list(self) > other def __ge__(self, other): return list(self) >= other def __cmp__(self, other): return util.cmp(list(self), other) def __add__(self, iterable): try: other = list(iterable) except TypeError: return NotImplemented return list(self) + other def __radd__(self, iterable): try: other = list(iterable) except TypeError: return NotImplemented return other + list(self) def __mul__(self, n): if not isinstance(n, int): return NotImplemented return list(self) n __rmul__ = __mul__ def __iadd__(self, iterable): self.extend(iterable) return self def __imul__(self, n): # unlike a regular list =, proxied __imul__ will generate unique # backing objects for each copy. = on proxied lists is a bit of # a stretch anyhow, and this interpretation of the __imul__ contract # is more plausibly useful than copying the backing objects. if not isinstance(n, int): return NotImplemented if n == 0: self.clear() elif n > 1: self.extend(list(self) * (n - 1)) return self def index(self, item, args): return list(self).index(item, args) def copy(self): return list(self) def __repr__(self): return repr(list(self)) def __hash__(self): raise TypeError("%s objects are unhashable" % type(self).__name__) for func_name, func in list(locals().items()): if ( util.callable(func) and func.__name__ == func_name and not func.__doc__ and hasattr(list, func_name) ): func.__doc__ = getattr(list, func_name).__doc__ del func_name, func _NotProvided = util.symbol("_NotProvided") class _AssociationDict(_AssociationCollection): """Generic, converting, dict-to-dict proxy.""" def _create(self, key, value): return self.creator(key, value) def _get(self, object_): return self.getter(object_) def _set(self, object_, key, value): return self.setter(object_, key, value) def __getitem__(self, key): return self._get(self.col[key]) def __setitem__(self, key, value): if key in self.col: self._set(self.col[key], key, value) else: self.col[key] = self._create(key, value) def __delitem__(self, key): del self.col[key] def __contains__(self, key): # testlib.pragma exempt:__hash__ return key in self.col def has_key(self, key): # testlib.pragma exempt:__hash__ return key in self.col def __iter__(self): return iter(self.col.keys()) def clear(self): self.col.clear() def __eq__(self, other): return dict(self) == other def __ne__(self, other): return dict(self) != other def __lt__(self, other): return dict(self) < other def __le__(self, other): return dict(self) <= other def __gt__(self, other): return dict(self) > other def __ge__(self, other): return dict(self) >= other def __cmp__(self, other): return util.cmp(dict(self), other) def __repr__(self): return repr(dict(self.items())) def get(self, key, default=None): try: return self[key] except KeyError: return default def setdefault(self, key, default=None): if key not in self.col: self.col[key] = self._create(key, default) return default else: return self[key] def keys(self): return self.col.keys() if util.py2k: def iteritems(self): return ((key, self._get(self.col[key])) for key in self.col) def itervalues(self): return (self._get(self.col[key]) for key in self.col) def iterkeys(self): return self.col.iterkeys() def values(self): return [self._get(member) for member in self.col.values()] def items(self): return [(k, self._get(self.col[k])) for k in self] else: def items(self): return ((key, self._get(self.col[key])) for key in self.col) def values(self): return (self._get(self.col[key]) for key in self.col) def pop(self, key, default=_NotProvided): if default is _NotProvided: member = self.col.pop(key) else: member = self.col.pop(key, default) return self._get(member) def popitem(self): item = self.col.popitem() return (item[0], self._get(item[1])) def update(self, a, *kw): if len(a) > 1: raise TypeError( "update expected at most 1 arguments, got %i" % len(a) ) elif len(a) == 1: seq_or_map = a[0] # discern dict from sequence - took the advice from # http://www.voidspace.org.uk/python/articles/duck_typing.shtml # still not perfect :( if hasattr(seq_or_map, "keys"): for item in seq_or_map: self[item] = seq_or_map[item] else: try: for k, v in seq_or_map: self[k] = v except ValueError as err: util.raise_( ValueError( "dictionary update sequence " "requires 2-element tuples" ), replace_context=err, ) for key, value in kw: self[key] = value def _bulk_replace(self, assoc_proxy, values): existing = set(self) constants = existing.intersection(values or ()) additions = set(values or ()).difference(constants) removals = existing.difference(constants) for key, member in values.items() or (): if key in additions: self[key] = member elif key in constants: self[key] = member for key in removals: del self[key] def copy(self): return dict(self.items()) def __hash__(self): raise TypeError("%s objects are unhashable" % type(self).__name__) for func_name, func in list(locals().items()): if ( util.callable(func) and func.__name__ == func_name and not func.__doc__ and hasattr(dict, func_name) ): func.__doc__ = getattr(dict, func_name).__doc__ del func_name, func class _AssociationSet(_AssociationCollection): """Generic, converting, set-to-set proxy.""" def _create(self, value): return self.creator(value) def _get(self, object_): return self.getter(object_) def __len__(self): return len(self.col) def __bool__(self): if self.col: return True else: return False __nonzero__ = __bool__ def __contains__(self, value): for member in self.col: # testlib.pragma exempt:__eq__ if self._get(member) == value: return True return False def __iter__(self): """Iterate over proxied values. For the actual domain objects, iterate over .col instead or just use the underlying collection directly from its property on the parent. """ for member in self.col: yield self._get(member) return def add(self, value): if value not in self: self.col.add(self._create(value)) # for discard and remove, choosing a more expensive check strategy rather # than call self.creator() def discard(self, value): for member in self.col: if self._get(member) == value: self.col.discard(member) break def remove(self, value): for member in self.col: if self._get(member) == value: self.col.discard(member) return raise KeyError(value) def pop(self): if not self.col: raise KeyError("pop from an empty set") member = self.col.pop() return self._get(member) def update(self, other): for value in other: self.add(value) def _bulk_replace(self, assoc_proxy, values): existing = set(self) constants = existing.intersection(values or ()) additions = set(values or ()).difference(constants) removals = existing.difference(constants) appender = self.add remover = self.remove for member in values or (): if member in additions: appender(member) elif member in constants: appender(member) for member in removals: remover(member) def __ior__(self, other): if not collections._set_binops_check_strict(self, other): return NotImplemented for value in other: self.add(value) return self def _set(self): return set(iter(self)) def union(self, other): return set(self).union(other) __or__ = union def difference(self, other): return set(self).difference(other) __sub__ = difference def difference_update(self, other): for value in other: self.discard(value) def __isub__(self, other): if not collections._set_binops_check_strict(self, other): return NotImplemented for value in other: self.discard(value) return self def intersection(self, other): return set(self).intersection(other) __and__ = intersection def intersection_update(self, other): want, have = self.intersection(other), set(self) remove, add = have - want, want - have for value in remove: self.remove(value) for value in add: self.add(value) def __iand__(self, other): if not collections._set_binops_check_strict(self, other): return NotImplemented want, have = self.intersection(other), set(self) remove, add = have - want, want - have for value in remove: self.remove(value) for value in add: self.add(value) return self def symmetric_difference(self, other): return set(self).symmetric_difference(other) __xor__ = symmetric_difference def symmetric_difference_update(self, other): want, have = self.symmetric_difference(other), set(self) remove, add = have - want, want - have for value in remove: self.remove(value) for value in add: self.add(value) def __ixor__(self, other): if not collections._set_binops_check_strict(self, other): return NotImplemented want, have = self.symmetric_difference(other), set(self) remove, add = have - want, want - have for value in remove: self.remove(value) for value in add: self.add(value) return self def issubset(self, other): return set(self).issubset(other) def issuperset(self, other): return set(self).issuperset(other) def clear(self): self.col.clear() def copy(self): return set(self) def __eq__(self, other): return set(self) == other def __ne__(self, other): return set(self) != other def __lt__(self, other): return set(self) < other def __le__(self, other): return set(self) <= other def __gt__(self, other): return set(self) > other def __ge__(self, other): return set(self) >= other def __repr__(self): return repr(set(self)) def __hash__(self): raise TypeError("%s objects are unhashable" % type(self).__name__) for func_name, func in list(locals().items()): if ( util.callable(func) and func.__name__ == func_name and not func.__doc__ and hasattr(set, func_name) ): func.__doc__ = getattr(set, func_name).__doc__ del func_name, func	48.81 KB	2026-01-20 12:58
📄 automap.py # ext/automap.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php r"""Define an extension to the :mod:`sqlalchemy.ext.declarative` system which automatically generates mapped classes and relationships from a database schema, typically though not necessarily one which is reflected. .. versionadded:: 0.9.1 Added :mod:`sqlalchemy.ext.automap`. It is hoped that the :class:`.AutomapBase` system provides a quick and modernized solution to the problem that the very famous `SQLSoup <https://sqlsoup.readthedocs.io/en/latest/>`_ also tries to solve, that of generating a quick and rudimentary object model from an existing database on the fly. By addressing the issue strictly at the mapper configuration level, and integrating fully with existing Declarative class techniques, :class:`.AutomapBase` seeks to provide a well-integrated approach to the issue of expediently auto-generating ad-hoc mappings. Basic Use ========= The simplest usage is to reflect an existing database into a new model. We create a new :class:`.AutomapBase` class in a similar manner as to how we create a declarative base class, using :func:`.automap_base`. We then call :meth:`.AutomapBase.prepare` on the resulting base class, asking it to reflect the schema and produce mappings:: from sqlalchemy.ext.automap import automap_base from sqlalchemy.orm import Session from sqlalchemy import create_engine Base = automap_base() # engine, suppose it has two tables 'user' and 'address' set up engine = create_engine("sqlite:///mydatabase.db") # reflect the tables Base.prepare(engine, reflect=True) # mapped classes are now created with names by default # matching that of the table name. User = Base.classes.user Address = Base.classes.address session = Session(engine) # rudimentary relationships are produced session.add(Address(email_address="foo@bar.com", user=User(name="foo"))) session.commit() # collection-based relationships are by default named # "<classname>_collection" print (u1.address_collection) Above, calling :meth:`.AutomapBase.prepare` while passing along the :paramref:`.AutomapBase.prepare.reflect` parameter indicates that the :meth:`_schema.MetaData.reflect` method will be called on this declarative base classes' :class:`_schema.MetaData` collection; then, each viable :class:`_schema.Table` within the :class:`_schema.MetaData` will get a new mapped class generated automatically. The :class:`_schema.ForeignKeyConstraint` objects which link the various tables together will be used to produce new, bidirectional :func:`_orm.relationship` objects between classes. The classes and relationships follow along a default naming scheme that we can customize. At this point, our basic mapping consisting of related ``User`` and ``Address`` classes is ready to use in the traditional way. .. note:: By viable, we mean that for a table to be mapped, it must specify a primary key. Additionally, if the table is detected as being a pure association table between two other tables, it will not be directly mapped and will instead be configured as a many-to-many table between the mappings for the two referring tables. Generating Mappings from an Existing MetaData ============================================= We can pass a pre-declared :class:`_schema.MetaData` object to :func:`.automap_base`. This object can be constructed in any way, including programmatically, from a serialized file, or from itself being reflected using :meth:`_schema.MetaData.reflect`. Below we illustrate a combination of reflection and explicit table declaration:: from sqlalchemy import create_engine, MetaData, Table, Column, ForeignKey from sqlalchemy.ext.automap import automap_base engine = create_engine("sqlite:///mydatabase.db") # produce our own MetaData object metadata = MetaData() # we can reflect it ourselves from a database, using options # such as 'only' to limit what tables we look at... metadata.reflect(engine, only=['user', 'address']) # ... or just define our own Table objects with it (or combine both) Table('user_order', metadata, Column('id', Integer, primary_key=True), Column('user_id', ForeignKey('user.id')) ) # we can then produce a set of mappings from this MetaData. Base = automap_base(metadata=metadata) # calling prepare() just sets up mapped classes and relationships. Base.prepare() # mapped classes are ready User, Address, Order = Base.classes.user, Base.classes.address,\ Base.classes.user_order Specifying Classes Explicitly ============================= The :mod:`.sqlalchemy.ext.automap` extension allows classes to be defined explicitly, in a way similar to that of the :class:`.DeferredReflection` class. Classes that extend from :class:`.AutomapBase` act like regular declarative classes, but are not immediately mapped after their construction, and are instead mapped when we call :meth:`.AutomapBase.prepare`. The :meth:`.AutomapBase.prepare` method will make use of the classes we've established based on the table name we use. If our schema contains tables ``user`` and ``address``, we can define one or both of the classes to be used:: from sqlalchemy.ext.automap import automap_base from sqlalchemy import create_engine # automap base Base = automap_base() # pre-declare User for the 'user' table class User(Base): __tablename__ = 'user' # override schema elements like Columns user_name = Column('name', String) # override relationships too, if desired. # we must use the same name that automap would use for the # relationship, and also must refer to the class name that automap will # generate for "address" address_collection = relationship("address", collection_class=set) # reflect engine = create_engine("sqlite:///mydatabase.db") Base.prepare(engine, reflect=True) # we still have Address generated from the tablename "address", # but User is the same as Base.classes.User now Address = Base.classes.address u1 = session.query(User).first() print (u1.address_collection) # the backref is still there: a1 = session.query(Address).first() print (a1.user) Above, one of the more intricate details is that we illustrated overriding one of the :func:`_orm.relationship` objects that automap would have created. To do this, we needed to make sure the names match up with what automap would normally generate, in that the relationship name would be ``User.address_collection`` and the name of the class referred to, from automap's perspective, is called ``address``, even though we are referring to it as ``Address`` within our usage of this class. Overriding Naming Schemes ========================= :mod:`.sqlalchemy.ext.automap` is tasked with producing mapped classes and relationship names based on a schema, which means it has decision points in how these names are determined. These three decision points are provided using functions which can be passed to the :meth:`.AutomapBase.prepare` method, and are known as :func:`.classname_for_table`, :func:`.name_for_scalar_relationship`, and :func:`.name_for_collection_relationship`. Any or all of these functions are provided as in the example below, where we use a "camel case" scheme for class names and a "pluralizer" for collection names using the `Inflect <https://pypi.python.org/pypi/inflect>`_ package:: import re import inflect def camelize_classname(base, tablename, table): "Produce a 'camelized' class name, e.g. " "'words_and_underscores' -> 'WordsAndUnderscores'" return str(tablename[0].upper() + \ re.sub(r'_([a-z])', lambda m: m.group(1).upper(), tablename[1:])) _pluralizer = inflect.engine() def pluralize_collection(base, local_cls, referred_cls, constraint): "Produce an 'uncamelized', 'pluralized' class name, e.g. " "'SomeTerm' -> 'some_terms'" referred_name = referred_cls.__name__ uncamelized = re.sub(r'[A-Z]', lambda m: "_%s" % m.group(0).lower(), referred_name)[1:] pluralized = _pluralizer.plural(uncamelized) return pluralized from sqlalchemy.ext.automap import automap_base Base = automap_base() engine = create_engine("sqlite:///mydatabase.db") Base.prepare(engine, reflect=True, classname_for_table=camelize_classname, name_for_collection_relationship=pluralize_collection ) From the above mapping, we would now have classes ``User`` and ``Address``, where the collection from ``User`` to ``Address`` is called ``User.addresses``:: User, Address = Base.classes.User, Base.classes.Address u1 = User(addresses=[Address(email="foo@bar.com")]) Relationship Detection ====================== The vast majority of what automap accomplishes is the generation of :func:`_orm.relationship` structures based on foreign keys. The mechanism by which this works for many-to-one and one-to-many relationships is as follows: 1. A given :class:`_schema.Table`, known to be mapped to a particular class, is examined for :class:`_schema.ForeignKeyConstraint` objects. 2. From each :class:`_schema.ForeignKeyConstraint`, the remote :class:`_schema.Table` object present is matched up to the class to which it is to be mapped, if any, else it is skipped. 3. As the :class:`_schema.ForeignKeyConstraint` we are examining corresponds to a reference from the immediate mapped class, the relationship will be set up as a many-to-one referring to the referred class; a corresponding one-to-many backref will be created on the referred class referring to this class. 4. If any of the columns that are part of the :class:`_schema.ForeignKeyConstraint` are not nullable (e.g. ``nullable=False``), a :paramref:`_orm.relationship.cascade` keyword argument of ``all, delete-orphan`` will be added to the keyword arguments to be passed to the relationship or backref. If the :class:`_schema.ForeignKeyConstraint` reports that :paramref:`_schema.ForeignKeyConstraint.ondelete` is set to ``CASCADE`` for a not null or ``SET NULL`` for a nullable set of columns, the option :paramref:`_orm.relationship.passive_deletes` flag is set to ``True`` in the set of relationship keyword arguments. Note that not all backends support reflection of ON DELETE. .. versionadded:: 1.0.0 - automap will detect non-nullable foreign key constraints when producing a one-to-many relationship and establish a default cascade of ``all, delete-orphan`` if so; additionally, if the constraint specifies :paramref:`_schema.ForeignKeyConstraint.ondelete` of ``CASCADE`` for non-nullable or ``SET NULL`` for nullable columns, the ``passive_deletes=True`` option is also added. 5. The names of the relationships are determined using the :paramref:`.AutomapBase.prepare.name_for_scalar_relationship` and :paramref:`.AutomapBase.prepare.name_for_collection_relationship` callable functions. It is important to note that the default relationship naming derives the name from the the actual class name. If you've given a particular class an explicit name by declaring it, or specified an alternate class naming scheme, that's the name from which the relationship name will be derived. 6. The classes are inspected for an existing mapped property matching these names. If one is detected on one side, but none on the other side, :class:`.AutomapBase` attempts to create a relationship on the missing side, then uses the :paramref:`_orm.relationship.back_populates` parameter in order to point the new relationship to the other side. 7. In the usual case where no relationship is on either side, :meth:`.AutomapBase.prepare` produces a :func:`_orm.relationship` on the "many-to-one" side and matches it to the other using the :paramref:`_orm.relationship.backref` parameter. 8. Production of the :func:`_orm.relationship` and optionally the :func:`.backref` is handed off to the :paramref:`.AutomapBase.prepare.generate_relationship` function, which can be supplied by the end-user in order to augment the arguments passed to :func:`_orm.relationship` or :func:`.backref` or to make use of custom implementations of these functions. Custom Relationship Arguments ----------------------------- The :paramref:`.AutomapBase.prepare.generate_relationship` hook can be used to add parameters to relationships. For most cases, we can make use of the existing :func:`.automap.generate_relationship` function to return the object, after augmenting the given keyword dictionary with our own arguments. Below is an illustration of how to send :paramref:`_orm.relationship.cascade` and :paramref:`_orm.relationship.passive_deletes` options along to all one-to-many relationships:: from sqlalchemy.ext.automap import generate_relationship def _gen_relationship(base, direction, return_fn, attrname, local_cls, referred_cls, kw): if direction is interfaces.ONETOMANY: kw['cascade'] = 'all, delete-orphan' kw['passive_deletes'] = True # make use of the built-in function to actually return # the result. return generate_relationship(base, direction, return_fn, attrname, local_cls, referred_cls, kw) from sqlalchemy.ext.automap import automap_base from sqlalchemy import create_engine # automap base Base = automap_base() engine = create_engine("sqlite:///mydatabase.db") Base.prepare(engine, reflect=True, generate_relationship=_gen_relationship) Many-to-Many relationships -------------------------- :mod:`.sqlalchemy.ext.automap` will generate many-to-many relationships, e.g. those which contain a ``secondary`` argument. The process for producing these is as follows: 1. A given :class:`_schema.Table` is examined for :class:`_schema.ForeignKeyConstraint` objects, before any mapped class has been assigned to it. 2. If the table contains two and exactly two :class:`_schema.ForeignKeyConstraint` objects, and all columns within this table are members of these two :class:`_schema.ForeignKeyConstraint` objects, the table is assumed to be a "secondary" table, and will not be mapped directly. 3. The two (or one, for self-referential) external tables to which the :class:`_schema.Table` refers to are matched to the classes to which they will be mapped, if any. 4. If mapped classes for both sides are located, a many-to-many bi-directional :func:`_orm.relationship` / :func:`.backref` pair is created between the two classes. 5. The override logic for many-to-many works the same as that of one-to-many/ many-to-one; the :func:`.generate_relationship` function is called upon to generate the structures and existing attributes will be maintained. Relationships with Inheritance ------------------------------ :mod:`.sqlalchemy.ext.automap` will not generate any relationships between two classes that are in an inheritance relationship. That is, with two classes given as follows:: class Employee(Base): __tablename__ = 'employee' id = Column(Integer, primary_key=True) type = Column(String(50)) __mapper_args__ = { 'polymorphic_identity':'employee', 'polymorphic_on': type } class Engineer(Employee): __tablename__ = 'engineer' id = Column(Integer, ForeignKey('employee.id'), primary_key=True) __mapper_args__ = { 'polymorphic_identity':'engineer', } The foreign key from ``Engineer`` to ``Employee`` is used not for a relationship, but to establish joined inheritance between the two classes. Note that this means automap will not generate any relationships for foreign keys that link from a subclass to a superclass. If a mapping has actual relationships from subclass to superclass as well, those need to be explicit. Below, as we have two separate foreign keys from ``Engineer`` to ``Employee``, we need to set up both the relationship we want as well as the ``inherit_condition``, as these are not things SQLAlchemy can guess:: class Employee(Base): __tablename__ = 'employee' id = Column(Integer, primary_key=True) type = Column(String(50)) __mapper_args__ = { 'polymorphic_identity':'employee', 'polymorphic_on':type } class Engineer(Employee): __tablename__ = 'engineer' id = Column(Integer, ForeignKey('employee.id'), primary_key=True) favorite_employee_id = Column(Integer, ForeignKey('employee.id')) favorite_employee = relationship(Employee, foreign_keys=favorite_employee_id) __mapper_args__ = { 'polymorphic_identity':'engineer', 'inherit_condition': id == Employee.id } Handling Simple Naming Conflicts -------------------------------- In the case of naming conflicts during mapping, override any of :func:`.classname_for_table`, :func:`.name_for_scalar_relationship`, and :func:`.name_for_collection_relationship` as needed. For example, if automap is attempting to name a many-to-one relationship the same as an existing column, an alternate convention can be conditionally selected. Given a schema: .. sourcecode:: sql CREATE TABLE table_a ( id INTEGER PRIMARY KEY ); CREATE TABLE table_b ( id INTEGER PRIMARY KEY, table_a INTEGER, FOREIGN KEY(table_a) REFERENCES table_a(id) ); The above schema will first automap the ``table_a`` table as a class named ``table_a``; it will then automap a relationship onto the class for ``table_b`` with the same name as this related class, e.g. ``table_a``. This relationship name conflicts with the mapping column ``table_b.table_a``, and will emit an error on mapping. We can resolve this conflict by using an underscore as follows:: def name_for_scalar_relationship(base, local_cls, referred_cls, constraint): name = referred_cls.__name__.lower() local_table = local_cls.__table__ if name in local_table.columns: newname = name + "_" warnings.warn( "Already detected name %s present. using %s" % (name, newname)) return newname return name Base.prepare(engine, reflect=True, name_for_scalar_relationship=name_for_scalar_relationship) Alternatively, we can change the name on the column side. The columns that are mapped can be modified using the technique described at :ref:`mapper_column_distinct_names`, by assigning the column explicitly to a new name:: Base = automap_base() class TableB(Base): __tablename__ = 'table_b' _table_a = Column('table_a', ForeignKey('table_a.id')) Base.prepare(engine, reflect=True) Using Automap with Explicit Declarations ======================================== As noted previously, automap has no dependency on reflection, and can make use of any collection of :class:`_schema.Table` objects within a :class:`_schema.MetaData` collection. From this, it follows that automap can also be used generate missing relationships given an otherwise complete model that fully defines table metadata:: from sqlalchemy.ext.automap import automap_base from sqlalchemy import Column, Integer, String, ForeignKey Base = automap_base() class User(Base): __tablename__ = 'user' id = Column(Integer, primary_key=True) name = Column(String) class Address(Base): __tablename__ = 'address' id = Column(Integer, primary_key=True) email = Column(String) user_id = Column(ForeignKey('user.id')) # produce relationships Base.prepare() # mapping is complete, with "address_collection" and # "user" relationships a1 = Address(email='u1') a2 = Address(email='u2') u1 = User(address_collection=[a1, a2]) assert a1.user is u1 Above, given mostly complete ``User`` and ``Address`` mappings, the :class:`_schema.ForeignKey` which we defined on ``Address.user_id`` allowed a bidirectional relationship pair ``Address.user`` and ``User.address_collection`` to be generated on the mapped classes. Note that when subclassing :class:`.AutomapBase`, the :meth:`.AutomapBase.prepare` method is required; if not called, the classes we've declared are in an un-mapped state. """ # noqa from .declarative import declarative_base as _declarative_base from .declarative.base import _DeferredMapperConfig from .. import util from ..orm import backref from ..orm import exc as orm_exc from ..orm import interfaces from ..orm import relationship from ..orm.mapper import _CONFIGURE_MUTEX from ..schema import ForeignKeyConstraint from ..sql import and_ def classname_for_table(base, tablename, table): """Return the class name that should be used, given the name of a table. The default implementation is:: return str(tablename) Alternate implementations can be specified using the :paramref:`.AutomapBase.prepare.classname_for_table` parameter. :param base: the :class:`.AutomapBase` class doing the prepare. :param tablename: string name of the :class:`_schema.Table`. :param table: the :class:`_schema.Table` object itself. :return: a string class name. .. note:: In Python 2, the string used for the class name must be a non-Unicode object, e.g. a ``str()`` object. The ``.name`` attribute of :class:`_schema.Table` is typically a Python unicode subclass, so the ``str()`` function should be applied to this name, after accounting for any non-ASCII characters. """ return str(tablename) def name_for_scalar_relationship(base, local_cls, referred_cls, constraint): """Return the attribute name that should be used to refer from one class to another, for a scalar object reference. The default implementation is:: return referred_cls.__name__.lower() Alternate implementations can be specified using the :paramref:`.AutomapBase.prepare.name_for_scalar_relationship` parameter. :param base: the :class:`.AutomapBase` class doing the prepare. :param local_cls: the class to be mapped on the local side. :param referred_cls: the class to be mapped on the referring side. :param constraint: the :class:`_schema.ForeignKeyConstraint` that is being inspected to produce this relationship. """ return referred_cls.__name__.lower() def name_for_collection_relationship( base, local_cls, referred_cls, constraint ): """Return the attribute name that should be used to refer from one class to another, for a collection reference. The default implementation is:: return referred_cls.__name__.lower() + "_collection" Alternate implementations can be specified using the :paramref:`.AutomapBase.prepare.name_for_collection_relationship` parameter. :param base: the :class:`.AutomapBase` class doing the prepare. :param local_cls: the class to be mapped on the local side. :param referred_cls: the class to be mapped on the referring side. :param constraint: the :class:`_schema.ForeignKeyConstraint` that is being inspected to produce this relationship. """ return referred_cls.__name__.lower() + "_collection" def generate_relationship( base, direction, return_fn, attrname, local_cls, referred_cls, kw ): r"""Generate a :func:`_orm.relationship` or :func:`.backref` on behalf of two mapped classes. An alternate implementation of this function can be specified using the :paramref:`.AutomapBase.prepare.generate_relationship` parameter. The default implementation of this function is as follows:: if return_fn is backref: return return_fn(attrname, kw) elif return_fn is relationship: return return_fn(referred_cls, kw) else: raise TypeError("Unknown relationship function: %s" % return_fn) :param base: the :class:`.AutomapBase` class doing the prepare. :param direction: indicate the "direction" of the relationship; this will be one of :data:`.ONETOMANY`, :data:`.MANYTOONE`, :data:`.MANYTOMANY`. :param return_fn: the function that is used by default to create the relationship. This will be either :func:`_orm.relationship` or :func:`.backref`. The :func:`.backref` function's result will be used to produce a new :func:`_orm.relationship` in a second step, so it is critical that user-defined implementations correctly differentiate between the two functions, if a custom relationship function is being used. :param attrname: the attribute name to which this relationship is being assigned. If the value of :paramref:`.generate_relationship.return_fn` is the :func:`.backref` function, then this name is the name that is being assigned to the backref. :param local_cls: the "local" class to which this relationship or backref will be locally present. :param referred_cls: the "referred" class to which the relationship or backref refers to. :param \kw: all additional keyword arguments are passed along to the function. :return: a :func:`_orm.relationship` or :func:`.backref` construct, as dictated by the :paramref:`.generate_relationship.return_fn` parameter. """ if return_fn is backref: return return_fn(attrname, kw) elif return_fn is relationship: return return_fn(referred_cls, kw) else: raise TypeError("Unknown relationship function: %s" % return_fn) class AutomapBase(object): """Base class for an "automap" schema. The :class:`.AutomapBase` class can be compared to the "declarative base" class that is produced by the :func:`.declarative.declarative_base` function. In practice, the :class:`.AutomapBase` class is always used as a mixin along with an actual declarative base. A new subclassable :class:`.AutomapBase` is typically instantiated using the :func:`.automap_base` function. .. seealso:: :ref:`automap_toplevel` """ __abstract__ = True classes = None """An instance of :class:`.util.Properties` containing classes. This object behaves much like the ``.c`` collection on a table. Classes are present under the name they were given, e.g.:: Base = automap_base() Base.prepare(engine=some_engine, reflect=True) User, Address = Base.classes.User, Base.classes.Address """ @classmethod def prepare( cls, engine=None, reflect=False, schema=None, classname_for_table=classname_for_table, collection_class=list, name_for_scalar_relationship=name_for_scalar_relationship, name_for_collection_relationship=name_for_collection_relationship, generate_relationship=generate_relationship, ): """Extract mapped classes and relationships from the :class:`_schema.MetaData` and perform mappings. :param engine: an :class:`_engine.Engine` or :class:`_engine.Connection` with which to perform schema reflection, if specified. If the :paramref:`.AutomapBase.prepare.reflect` argument is False, this object is not used. :param reflect: if True, the :meth:`_schema.MetaData.reflect` method is called on the :class:`_schema.MetaData` associated with this :class:`.AutomapBase`. The :class:`_engine.Engine` passed via :paramref:`.AutomapBase.prepare.engine` will be used to perform the reflection if present; else, the :class:`_schema.MetaData` should already be bound to some engine else the operation will fail. :param classname_for_table: callable function which will be used to produce new class names, given a table name. Defaults to :func:`.classname_for_table`. :param name_for_scalar_relationship: callable function which will be used to produce relationship names for scalar relationships. Defaults to :func:`.name_for_scalar_relationship`. :param name_for_collection_relationship: callable function which will be used to produce relationship names for collection-oriented relationships. Defaults to :func:`.name_for_collection_relationship`. :param generate_relationship: callable function which will be used to actually generate :func:`_orm.relationship` and :func:`.backref` constructs. Defaults to :func:`.generate_relationship`. :param collection_class: the Python collection class that will be used when a new :func:`_orm.relationship` object is created that represents a collection. Defaults to ``list``. :param schema: When present in conjunction with the :paramref:`.AutomapBase.prepare.reflect` flag, is passed to :meth:`_schema.MetaData.reflect` to indicate the primary schema where tables should be reflected from. When omitted, the default schema in use by the database connection is used. .. versionadded:: 1.1 """ if reflect: cls.metadata.reflect( engine, schema=schema, extend_existing=True, autoload_replace=False, ) _CONFIGURE_MUTEX.acquire() try: table_to_map_config = dict( (m.local_table, m) for m in _DeferredMapperConfig.classes_for_base( cls, sort=False ) ) many_to_many = [] for table in cls.metadata.tables.values(): lcl_m2m, rem_m2m, m2m_const = _is_many_to_many(cls, table) if lcl_m2m is not None: many_to_many.append((lcl_m2m, rem_m2m, m2m_const, table)) elif not table.primary_key: continue elif table not in table_to_map_config: mapped_cls = type( classname_for_table(cls, table.name, table), (cls,), {"__table__": table}, ) map_config = _DeferredMapperConfig.config_for_cls( mapped_cls ) cls.classes[map_config.cls.__name__] = mapped_cls table_to_map_config[table] = map_config for map_config in table_to_map_config.values(): _relationships_for_fks( cls, map_config, table_to_map_config, collection_class, name_for_scalar_relationship, name_for_collection_relationship, generate_relationship, ) for lcl_m2m, rem_m2m, m2m_const, table in many_to_many: _m2m_relationship( cls, lcl_m2m, rem_m2m, m2m_const, table, table_to_map_config, collection_class, name_for_scalar_relationship, name_for_collection_relationship, generate_relationship, ) for map_config in _DeferredMapperConfig.classes_for_base(cls): map_config.map() finally: _CONFIGURE_MUTEX.release() _sa_decl_prepare = True """Indicate that the mapping of classes should be deferred. The presence of this attribute name indicates to declarative that the call to mapper() should not occur immediately; instead, information about the table and attributes to be mapped are gathered into an internal structure called _DeferredMapperConfig. These objects can be collected later using classes_for_base(), additional mapping decisions can be made, and then the map() method will actually apply the mapping. The only real reason this deferral of the whole thing is needed is to support primary key columns that aren't reflected yet when the class is declared; everything else can theoretically be added to the mapper later. However, the _DeferredMapperConfig is a nice interface in any case which exists at that not usually exposed point at which declarative has the class and the Table but hasn't called mapper() yet. """ @classmethod def _sa_raise_deferred_config(cls): raise orm_exc.UnmappedClassError( cls, msg="Class %s is a subclass of AutomapBase. " "Mappings are not produced until the .prepare() " "method is called on the class hierarchy." % orm_exc._safe_cls_name(cls), ) def automap_base(declarative_base=None, kw): r"""Produce a declarative automap base. This function produces a new base class that is a product of the :class:`.AutomapBase` class as well a declarative base produced by :func:`.declarative.declarative_base`. All parameters other than ``declarative_base`` are keyword arguments that are passed directly to the :func:`.declarative.declarative_base` function. :param declarative_base: an existing class produced by :func:`.declarative.declarative_base`. When this is passed, the function no longer invokes :func:`.declarative.declarative_base` itself, and all other keyword arguments are ignored. :param \kw: keyword arguments are passed along to :func:`.declarative.declarative_base`. """ if declarative_base is None: Base = _declarative_base(kw) else: Base = declarative_base return type( Base.__name__, (AutomapBase, Base), {"__abstract__": True, "classes": util.Properties({})}, ) def _is_many_to_many(automap_base, table): fk_constraints = [ const for const in table.constraints if isinstance(const, ForeignKeyConstraint) ] if len(fk_constraints) != 2: return None, None, None cols = sum( [ [fk.parent for fk in fk_constraint.elements] for fk_constraint in fk_constraints ], [], ) if set(cols) != set(table.c): return None, None, None return ( fk_constraints[0].elements[0].column.table, fk_constraints[1].elements[0].column.table, fk_constraints, ) def _relationships_for_fks( automap_base, map_config, table_to_map_config, collection_class, name_for_scalar_relationship, name_for_collection_relationship, generate_relationship, ): local_table = map_config.local_table local_cls = map_config.cls # derived from a weakref, may be None if local_table is None or local_cls is None: return for constraint in local_table.constraints: if isinstance(constraint, ForeignKeyConstraint): fks = constraint.elements referred_table = fks[0].column.table referred_cfg = table_to_map_config.get(referred_table, None) if referred_cfg is None: continue referred_cls = referred_cfg.cls if local_cls is not referred_cls and issubclass( local_cls, referred_cls ): continue relationship_name = name_for_scalar_relationship( automap_base, local_cls, referred_cls, constraint ) backref_name = name_for_collection_relationship( automap_base, referred_cls, local_cls, constraint ) o2m_kws = {} nullable = False not in {fk.parent.nullable for fk in fks} if not nullable: o2m_kws["cascade"] = "all, delete-orphan" if ( constraint.ondelete and constraint.ondelete.lower() == "cascade" ): o2m_kws["passive_deletes"] = True else: if ( constraint.ondelete and constraint.ondelete.lower() == "set null" ): o2m_kws["passive_deletes"] = True create_backref = backref_name not in referred_cfg.properties if relationship_name not in map_config.properties: if create_backref: backref_obj = generate_relationship( automap_base, interfaces.ONETOMANY, backref, backref_name, referred_cls, local_cls, collection_class=collection_class, o2m_kws ) else: backref_obj = None rel = generate_relationship( automap_base, interfaces.MANYTOONE, relationship, relationship_name, local_cls, referred_cls, foreign_keys=[fk.parent for fk in constraint.elements], backref=backref_obj, remote_side=[fk.column for fk in constraint.elements], ) if rel is not None: map_config.properties[relationship_name] = rel if not create_backref: referred_cfg.properties[ backref_name ].back_populates = relationship_name elif create_backref: rel = generate_relationship( automap_base, interfaces.ONETOMANY, relationship, backref_name, referred_cls, local_cls, foreign_keys=[fk.parent for fk in constraint.elements], back_populates=relationship_name, collection_class=collection_class, **o2m_kws ) if rel is not None: referred_cfg.properties[backref_name] = rel map_config.properties[ relationship_name ].back_populates = backref_name def _m2m_relationship( automap_base, lcl_m2m, rem_m2m, m2m_const, table, table_to_map_config, collection_class, name_for_scalar_relationship, name_for_collection_relationship, generate_relationship, ): map_config = table_to_map_config.get(lcl_m2m, None) referred_cfg = table_to_map_config.get(rem_m2m, None) if map_config is None or referred_cfg is None: return local_cls = map_config.cls referred_cls = referred_cfg.cls relationship_name = name_for_collection_relationship( automap_base, local_cls, referred_cls, m2m_const[0] ) backref_name = name_for_collection_relationship( automap_base, referred_cls, local_cls, m2m_const[1] ) create_backref = backref_name not in referred_cfg.properties if relationship_name not in map_config.properties: if create_backref: backref_obj = generate_relationship( automap_base, interfaces.MANYTOMANY, backref, backref_name, referred_cls, local_cls, collection_class=collection_class, ) else: backref_obj = None rel = generate_relationship( automap_base, interfaces.MANYTOMANY, relationship, relationship_name, local_cls, referred_cls, secondary=table, primaryjoin=and_( fk.column == fk.parent for fk in m2m_const[0].elements ), secondaryjoin=and_( fk.column == fk.parent for fk in m2m_const[1].elements ), backref=backref_obj, collection_class=collection_class, ) if rel is not None: map_config.properties[relationship_name] = rel if not create_backref: referred_cfg.properties[ backref_name ].back_populates = relationship_name elif create_backref: rel = generate_relationship( automap_base, interfaces.MANYTOMANY, relationship, backref_name, referred_cls, local_cls, secondary=table, primaryjoin=and_( fk.column == fk.parent for fk in m2m_const[1].elements ), secondaryjoin=and_( fk.column == fk.parent for fk in m2m_const[0].elements ), back_populates=relationship_name, collection_class=collection_class, ) if rel is not None: referred_cfg.properties[backref_name] = rel map_config.properties[ relationship_name ].back_populates = backref_name	41.17 KB	2026-01-20 12:58
📄 baked.py # sqlalchemy/ext/baked.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Baked query extension. Provides a creational pattern for the :class:`.query.Query` object which allows the fully constructed object, Core select statement, and string compiled result to be fully cached. """ import copy import logging from .. import exc as sa_exc from .. import util from ..orm import exc as orm_exc from ..orm import strategy_options from ..orm.query import Query from ..orm.session import Session from ..sql import func from ..sql import literal_column from ..sql import util as sql_util log = logging.getLogger(__name__) class Bakery(object): """Callable which returns a :class:`.BakedQuery`. This object is returned by the class method :meth:`.BakedQuery.bakery`. It exists as an object so that the "cache" can be easily inspected. .. versionadded:: 1.2 """ __slots__ = "cls", "cache" def __init__(self, cls_, cache): self.cls = cls_ self.cache = cache def __call__(self, initial_fn, args): return self.cls(self.cache, initial_fn, args) class BakedQuery(object): """A builder object for :class:`.query.Query` objects.""" __slots__ = "steps", "_bakery", "_cache_key", "_spoiled" def __init__(self, bakery, initial_fn, args=()): self._cache_key = () self._update_cache_key(initial_fn, args) self.steps = [initial_fn] self._spoiled = False self._bakery = bakery @classmethod def bakery(cls, size=200, _size_alert=None): """Construct a new bakery. :return: an instance of :class:`.Bakery` """ return Bakery(cls, util.LRUCache(size, size_alert=_size_alert)) def _clone(self): b1 = BakedQuery.__new__(BakedQuery) b1._cache_key = self._cache_key b1.steps = list(self.steps) b1._bakery = self._bakery b1._spoiled = self._spoiled return b1 def _update_cache_key(self, fn, args=()): self._cache_key += (fn.__code__,) + args def __iadd__(self, other): if isinstance(other, tuple): self.add_criteria(other) else: self.add_criteria(other) return self def __add__(self, other): if isinstance(other, tuple): return self.with_criteria(other) else: return self.with_criteria(other) def add_criteria(self, fn, args): """Add a criteria function to this :class:`.BakedQuery`. This is equivalent to using the ``+=`` operator to modify a :class:`.BakedQuery` in-place. """ self._update_cache_key(fn, args) self.steps.append(fn) return self def with_criteria(self, fn, args): """Add a criteria function to a :class:`.BakedQuery` cloned from this one. This is equivalent to using the ``+`` operator to produce a new :class:`.BakedQuery` with modifications. """ return self._clone().add_criteria(fn, args) def for_session(self, session): """Return a :class:`.Result` object for this :class:`.BakedQuery`. This is equivalent to calling the :class:`.BakedQuery` as a Python callable, e.g. ``result = my_baked_query(session)``. """ return Result(self, session) def __call__(self, session): return self.for_session(session) def spoil(self, full=False): """Cancel any query caching that will occur on this BakedQuery object. The BakedQuery can continue to be used normally, however additional creational functions will not be cached; they will be called on every invocation. This is to support the case where a particular step in constructing a baked query disqualifies the query from being cacheable, such as a variant that relies upon some uncacheable value. :param full: if False, only functions added to this :class:`.BakedQuery` object subsequent to the spoil step will be non-cached; the state of the :class:`.BakedQuery` up until this point will be pulled from the cache. If True, then the entire :class:`_query.Query` object is built from scratch each time, with all creational functions being called on each invocation. """ if not full and not self._spoiled: _spoil_point = self._clone() _spoil_point._cache_key += ("_query_only",) self.steps = [_spoil_point._retrieve_baked_query] self._spoiled = True return self def _effective_key(self, session): """Return the key that actually goes into the cache dictionary for this :class:`.BakedQuery`, taking into account the given :class:`.Session`. This basically means we also will include the session's query_class, as the actual :class:`_query.Query` object is part of what's cached and needs to match the type of :class:`_query.Query` that a later session will want to use. """ return self._cache_key + (session._query_cls,) def _with_lazyload_options(self, options, effective_path, cache_path=None): """Cloning version of _add_lazyload_options.""" q = self._clone() q._add_lazyload_options(options, effective_path, cache_path=cache_path) return q def _add_lazyload_options(self, options, effective_path, cache_path=None): """Used by per-state lazy loaders to add options to the "lazy load" query from a parent query. Creates a cache key based on given load path and query options; if a repeatable cache key cannot be generated, the query is "spoiled" so that it won't use caching. """ key = () if not cache_path: cache_path = effective_path if cache_path.path[0].is_aliased_class: # paths that are against an AliasedClass are unsafe to cache # with since the AliasedClass is an ad-hoc object. self.spoil(full=True) else: for opt in options: cache_key = opt._generate_cache_key(cache_path) if cache_key is False: self.spoil(full=True) elif cache_key is not None: key += cache_key self.add_criteria( lambda q: q._with_current_path( effective_path )._conditional_options(options), cache_path.path, key, ) def _retrieve_baked_query(self, session): query = self._bakery.get(self._effective_key(session), None) if query is None: query = self._as_query(session) self._bakery[self._effective_key(session)] = query.with_session( None ) return query.with_session(session) def _bake(self, session): query = self._as_query(session) context = query._compile_context() self._bake_subquery_loaders(session, context) context.session = None context.query = query = context.query.with_session(None) query._execution_options = query._execution_options.union( {"compiled_cache": self._bakery} ) # we'll be holding onto the query for some of its state, # so delete some compilation-use-only attributes that can take up # space for attr in ( "_correlate", "_from_obj", "_mapper_adapter_map", "_joinpath", "_joinpoint", ): query.__dict__.pop(attr, None) # if the query is not safe to cache, we still do everything as though # we did cache it, since the receiver of _bake() assumes subqueryload # context was set up, etc. if context.query._bake_ok: self._bakery[self._effective_key(session)] = context return context def to_query(self, query_or_session): """Return the :class:`_query.Query` object for use as a subquery. This method should be used within the lambda callable being used to generate a step of an enclosing :class:`.BakedQuery`. The parameter should normally be the :class:`_query.Query` object that is passed to the lambda:: sub_bq = self.bakery(lambda s: s.query(User.name)) sub_bq += lambda q: q.filter( User.id == Address.user_id).correlate(Address) main_bq = self.bakery(lambda s: s.query(Address)) main_bq += lambda q: q.filter( sub_bq.to_query(q).exists()) In the case where the subquery is used in the first callable against a :class:`.Session`, the :class:`.Session` is also accepted:: sub_bq = self.bakery(lambda s: s.query(User.name)) sub_bq += lambda q: q.filter( User.id == Address.user_id).correlate(Address) main_bq = self.bakery( lambda s: s.query(Address.id, sub_bq.to_query(q).as_scalar()) ) :param query_or_session: a :class:`_query.Query` object or a class :class:`.Session` object, that is assumed to be within the context of an enclosing :class:`.BakedQuery` callable. .. versionadded:: 1.3 """ if isinstance(query_or_session, Session): session = query_or_session elif isinstance(query_or_session, Query): session = query_or_session.session if session is None: raise sa_exc.ArgumentError( "Given Query needs to be associated with a Session" ) else: raise TypeError( "Query or Session object expected, got %r." % type(query_or_session) ) return self._as_query(session) def _as_query(self, session): query = self.steps[0](session) for step in self.steps[1:]: query = step(query) return query def _bake_subquery_loaders(self, session, context): """convert subquery eager loaders in the cache into baked queries. For subquery eager loading to work, all we need here is that the Query point to the correct session when it is run. However, since we are "baking" anyway, we may as well also turn the query into a "baked" query so that we save on performance too. """ context.attributes["baked_queries"] = baked_queries = [] for k, v in list(context.attributes.items()): if isinstance(v, Query): if "subquery" in k: bk = BakedQuery(self._bakery, lambda args: v) bk._cache_key = self._cache_key + k bk._bake(session) baked_queries.append((k, bk._cache_key, v)) del context.attributes[k] def _unbake_subquery_loaders( self, session, context, params, post_criteria ): """Retrieve subquery eager loaders stored by _bake_subquery_loaders and turn them back into Result objects that will iterate just like a Query object. """ if "baked_queries" not in context.attributes: return for k, cache_key, query in context.attributes["baked_queries"]: bk = BakedQuery( self._bakery, lambda sess, q=query: q.with_session(sess) ) bk._cache_key = cache_key q = bk.for_session(session) for fn in post_criteria: q = q.with_post_criteria(fn) context.attributes[k] = q.params(*params) class Result(object): """Invokes a :class:`.BakedQuery` against a :class:`.Session`. The :class:`.Result` object is where the actual :class:`.query.Query` object gets created, or retrieved from the cache, against a target :class:`.Session`, and is then invoked for results. """ __slots__ = "bq", "session", "_params", "_post_criteria" def __init__(self, bq, session): self.bq = bq self.session = session self._params = {} self._post_criteria = [] def params(self, args, kw): """Specify parameters to be replaced into the string SQL statement.""" if len(args) == 1: kw.update(args[0]) elif len(args) > 0: raise sa_exc.ArgumentError( "params() takes zero or one positional argument, " "which is a dictionary." ) self._params.update(kw) return self def _using_post_criteria(self, fns): if fns: self._post_criteria.extend(fns) return self def with_post_criteria(self, fn): """Add a criteria function that will be applied post-cache. This adds a function that will be run against the :class:`_query.Query` object after it is retrieved from the cache. Functions here can be used to alter the query in ways that do not affect the SQL output, such as execution options and shard identifiers (when using a shard-enabled query object) .. warning:: :meth:`.Result.with_post_criteria` functions are applied to the :class:`_query.Query` object after** the query's SQL statement object has been retrieved from the cache. Any operations here which intend to modify the SQL should ensure that :meth:`.BakedQuery.spoil` was called first. .. versionadded:: 1.2 """ return self._using_post_criteria([fn]) def _as_query(self): q = self.bq._as_query(self.session).params(self._params) for fn in self._post_criteria: q = fn(q) return q def __str__(self): return str(self._as_query()) def __iter__(self): bq = self.bq if not self.session.enable_baked_queries or bq._spoiled: return iter(self._as_query()) baked_context = bq._bakery.get(bq._effective_key(self.session), None) if baked_context is None: baked_context = bq._bake(self.session) context = copy.copy(baked_context) context.session = self.session context.attributes = context.attributes.copy() bq._unbake_subquery_loaders( self.session, context, self._params, self._post_criteria ) context.statement.use_labels = True if context.autoflush and not context.populate_existing: self.session._autoflush() q = context.query.params(self._params).with_session(self.session) for fn in self._post_criteria: q = fn(q) return q._execute_and_instances(context) def count(self): """return the 'count'. Equivalent to :meth:`_query.Query.count`. Note this uses a subquery to ensure an accurate count regardless of the structure of the original statement. .. versionadded:: 1.1.6 """ col = func.count(literal_column("")) bq = self.bq.with_criteria(lambda q: q.from_self(col)) return bq.for_session(self.session).params(self._params).scalar() def scalar(self): """Return the first element of the first result or None if no rows present. If multiple rows are returned, raises MultipleResultsFound. Equivalent to :meth:`_query.Query.scalar`. .. versionadded:: 1.1.6 """ try: ret = self.one() if not isinstance(ret, tuple): return ret return ret[0] except orm_exc.NoResultFound: return None def first(self): """Return the first row. Equivalent to :meth:`_query.Query.first`. """ bq = self.bq.with_criteria(lambda q: q.slice(0, 1)) ret = list( bq.for_session(self.session) .params(self._params) ._using_post_criteria(self._post_criteria) ) if len(ret) > 0: return ret[0] else: return None def one(self): """Return exactly one result or raise an exception. Equivalent to :meth:`_query.Query.one`. """ try: ret = self.one_or_none() except orm_exc.MultipleResultsFound as err: util.raise_( orm_exc.MultipleResultsFound( "Multiple rows were found for one()" ), replace_context=err, ) else: if ret is None: raise orm_exc.NoResultFound("No row was found for one()") return ret def one_or_none(self): """Return one or zero results, or raise an exception for multiple rows. Equivalent to :meth:`_query.Query.one_or_none`. .. versionadded:: 1.0.9 """ ret = list(self) l = len(ret) if l == 1: return ret[0] elif l == 0: return None else: raise orm_exc.MultipleResultsFound( "Multiple rows were found for one_or_none()" ) def all(self): """Return all rows. Equivalent to :meth:`_query.Query.all`. """ return list(self) def get(self, ident): """Retrieve an object based on identity. Equivalent to :meth:`_query.Query.get`. """ query = self.bq.steps[0](self.session) return query._get_impl(ident, self._load_on_pk_identity) def _load_on_pk_identity(self, query, primary_key_identity): """Load the given primary key identity from the database.""" mapper = query._mapper_zero() _get_clause, _get_params = mapper._get_clause def setup(query): _lcl_get_clause = _get_clause q = query._clone() q._get_condition() q._order_by = None # None present in ident - turn those comparisons # into "IS NULL" if None in primary_key_identity: nones = set( [ _get_params[col].key for col, value in zip( mapper.primary_key, primary_key_identity ) if value is None ] ) _lcl_get_clause = sql_util.adapt_criterion_to_null( _lcl_get_clause, nones ) _lcl_get_clause = q._adapt_clause(_lcl_get_clause, True, False) q._criterion = _lcl_get_clause for fn in self._post_criteria: q = fn(q) return q # cache the query against a key that includes # which positions in the primary key are NULL # (remember, we can map to an OUTER JOIN) bq = self.bq # add the clause we got from mapper._get_clause to the cache # key so that if a race causes multiple calls to _get_clause, # we've cached on ours bq = bq._clone() bq._cache_key += (_get_clause,) bq = bq.with_criteria( setup, tuple(elem is None for elem in primary_key_identity) ) params = dict( [ (_get_params[primary_key].key, id_val) for id_val, primary_key in zip( primary_key_identity, mapper.primary_key ) ] ) result = list(bq.for_session(self.session).params(params)) l = len(result) if l > 1: raise orm_exc.MultipleResultsFound() elif l: return result[0] else: return None @util.deprecated( "1.2", "Baked lazy loading is now the default implementation." ) def bake_lazy_loaders(): """Enable the use of baked queries for all lazyloaders systemwide. The "baked" implementation of lazy loading is now the sole implementation for the base lazy loader; this method has no effect except for a warning. """ pass @util.deprecated( "1.2", "Baked lazy loading is now the default implementation." ) def unbake_lazy_loaders(): """Disable the use of baked queries for all lazyloaders systemwide. This method now raises NotImplementedError() as the "baked" implementation is the only lazy load implementation. The :paramref:`_orm.relationship.bake_queries` flag may be used to disable the caching of queries on a per-relationship basis. """ raise NotImplementedError( "Baked lazy loading is now the default implementation" ) @strategy_options.loader_option() def baked_lazyload(loadopt, attr): """Indicate that the given attribute should be loaded using "lazy" loading with a "baked" query used in the load. """ return loadopt.set_relationship_strategy(attr, {"lazy": "baked_select"}) @baked_lazyload._add_unbound_fn @util.deprecated( "1.2", "Baked lazy loading is now the default " "implementation for lazy loading.", ) def baked_lazyload(keys): return strategy_options._UnboundLoad._from_keys( strategy_options._UnboundLoad.baked_lazyload, keys, False, {} ) @baked_lazyload._add_unbound_all_fn @util.deprecated( "1.2", "Baked lazy loading is now the default " "implementation for lazy loading.", ) def baked_lazyload_all(*keys): return strategy_options._UnboundLoad._from_keys( strategy_options._UnboundLoad.baked_lazyload, keys, True, {} ) baked_lazyload = baked_lazyload._unbound_fn baked_lazyload_all = baked_lazyload_all._unbound_all_fn bakery = BakedQuery.bakery	21.46 KB	2026-01-20 12:58
📄 compiler.py # ext/compiler.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php r"""Provides an API for creation of custom ClauseElements and compilers. Synopsis ======== Usage involves the creation of one or more :class:`~sqlalchemy.sql.expression.ClauseElement` subclasses and one or more callables defining its compilation:: from sqlalchemy.ext.compiler import compiles from sqlalchemy.sql.expression import ColumnClause class MyColumn(ColumnClause): pass @compiles(MyColumn) def compile_mycolumn(element, compiler, kw): return "[%s]" % element.name Above, ``MyColumn`` extends :class:`~sqlalchemy.sql.expression.ColumnClause`, the base expression element for named column objects. The ``compiles`` decorator registers itself with the ``MyColumn`` class so that it is invoked when the object is compiled to a string:: from sqlalchemy import select s = select([MyColumn('x'), MyColumn('y')]) print(str(s)) Produces:: SELECT [x], [y] Dialect-specific compilation rules ================================== Compilers can also be made dialect-specific. The appropriate compiler will be invoked for the dialect in use:: from sqlalchemy.schema import DDLElement class AlterColumn(DDLElement): def __init__(self, column, cmd): self.column = column self.cmd = cmd @compiles(AlterColumn) def visit_alter_column(element, compiler, kw): return "ALTER COLUMN %s ..." % element.column.name @compiles(AlterColumn, 'postgresql') def visit_alter_column(element, compiler, kw): return "ALTER TABLE %s ALTER COLUMN %s ..." % (element.table.name, element.column.name) The second ``visit_alter_table`` will be invoked when any ``postgresql`` dialect is used. Compiling sub-elements of a custom expression construct ======================================================= The ``compiler`` argument is the :class:`~sqlalchemy.engine.interfaces.Compiled` object in use. This object can be inspected for any information about the in-progress compilation, including ``compiler.dialect``, ``compiler.statement`` etc. The :class:`~sqlalchemy.sql.compiler.SQLCompiler` and :class:`~sqlalchemy.sql.compiler.DDLCompiler` both include a ``process()`` method which can be used for compilation of embedded attributes:: from sqlalchemy.sql.expression import Executable, ClauseElement class InsertFromSelect(Executable, ClauseElement): def __init__(self, table, select): self.table = table self.select = select @compiles(InsertFromSelect) def visit_insert_from_select(element, compiler, kw): return "INSERT INTO %s (%s)" % ( compiler.process(element.table, asfrom=True, kw), compiler.process(element.select, kw) ) insert = InsertFromSelect(t1, select([t1]).where(t1.c.x>5)) print(insert) Produces:: "INSERT INTO mytable (SELECT mytable.x, mytable.y, mytable.z FROM mytable WHERE mytable.x > :x_1)" .. note:: The above ``InsertFromSelect`` construct is only an example, this actual functionality is already available using the :meth:`_expression.Insert.from_select` method. .. note:: The above ``InsertFromSelect`` construct probably wants to have "autocommit" enabled. See :ref:`enabling_compiled_autocommit` for this step. Cross Compiling between SQL and DDL compilers --------------------------------------------- SQL and DDL constructs are each compiled using different base compilers - ``SQLCompiler`` and ``DDLCompiler``. A common need is to access the compilation rules of SQL expressions from within a DDL expression. The ``DDLCompiler`` includes an accessor ``sql_compiler`` for this reason, such as below where we generate a CHECK constraint that embeds a SQL expression:: @compiles(MyConstraint) def compile_my_constraint(constraint, ddlcompiler, kw): kw['literal_binds'] = True return "CONSTRAINT %s CHECK (%s)" % ( constraint.name, ddlcompiler.sql_compiler.process( constraint.expression, kw) ) Above, we add an additional flag to the process step as called by :meth:`.SQLCompiler.process`, which is the ``literal_binds`` flag. This indicates that any SQL expression which refers to a :class:`.BindParameter` object or other "literal" object such as those which refer to strings or integers should be rendered in-place, rather than being referred to as a bound parameter; when emitting DDL, bound parameters are typically not supported. .. _enabling_compiled_autocommit: Enabling Autocommit on a Construct ================================== Recall from the section :ref:`autocommit` that the :class:`_engine.Engine`, when asked to execute a construct in the absence of a user-defined transaction, detects if the given construct represents DML or DDL, that is, a data modification or data definition statement, which requires (or may require, in the case of DDL) that the transaction generated by the DBAPI be committed (recall that DBAPI always has a transaction going on regardless of what SQLAlchemy does). Checking for this is actually accomplished by checking for the "autocommit" execution option on the construct. When building a construct like an INSERT derivation, a new DDL type, or perhaps a stored procedure that alters data, the "autocommit" option needs to be set in order for the statement to function with "connectionless" execution (as described in :ref:`dbengine_implicit`). Currently a quick way to do this is to subclass :class:`.Executable`, then add the "autocommit" flag to the ``_execution_options`` dictionary (note this is a "frozen" dictionary which supplies a generative ``union()`` method):: from sqlalchemy.sql.expression import Executable, ClauseElement class MyInsertThing(Executable, ClauseElement): _execution_options = \ Executable._execution_options.union({'autocommit': True}) More succinctly, if the construct is truly similar to an INSERT, UPDATE, or DELETE, :class:`.UpdateBase` can be used, which already is a subclass of :class:`.Executable`, :class:`_expression.ClauseElement` and includes the ``autocommit`` flag:: from sqlalchemy.sql.expression import UpdateBase class MyInsertThing(UpdateBase): def __init__(self, ...): ... DDL elements that subclass :class:`.DDLElement` already have the "autocommit" flag turned on. Changing the default compilation of existing constructs ======================================================= The compiler extension applies just as well to the existing constructs. When overriding the compilation of a built in SQL construct, the @compiles decorator is invoked upon the appropriate class (be sure to use the class, i.e. ``Insert`` or ``Select``, instead of the creation function such as ``insert()`` or ``select()``). Within the new compilation function, to get at the "original" compilation routine, use the appropriate visit_XXX method - this because compiler.process() will call upon the overriding routine and cause an endless loop. Such as, to add "prefix" to all insert statements:: from sqlalchemy.sql.expression import Insert @compiles(Insert) def prefix_inserts(insert, compiler, kw): return compiler.visit_insert(insert.prefix_with("some prefix"), kw) The above compiler will prefix all INSERT statements with "some prefix" when compiled. .. _type_compilation_extension: Changing Compilation of Types ============================= ``compiler`` works for types, too, such as below where we implement the MS-SQL specific 'max' keyword for ``String``/``VARCHAR``:: @compiles(String, 'mssql') @compiles(VARCHAR, 'mssql') def compile_varchar(element, compiler, kw): if element.length == 'max': return "VARCHAR('max')" else: return compiler.visit_VARCHAR(element, kw) foo = Table('foo', metadata, Column('data', VARCHAR('max')) ) Subclassing Guidelines ====================== A big part of using the compiler extension is subclassing SQLAlchemy expression constructs. To make this easier, the expression and schema packages feature a set of "bases" intended for common tasks. A synopsis is as follows: * :class:`~sqlalchemy.sql.expression.ClauseElement` - This is the root expression class. Any SQL expression can be derived from this base, and is probably the best choice for longer constructs such as specialized INSERT statements. * :class:`~sqlalchemy.sql.expression.ColumnElement` - The root of all "column-like" elements. Anything that you'd place in the "columns" clause of a SELECT statement (as well as order by and group by) can derive from this - the object will automatically have Python "comparison" behavior. :class:`~sqlalchemy.sql.expression.ColumnElement` classes want to have a ``type`` member which is expression's return type. This can be established at the instance level in the constructor, or at the class level if its generally constant:: class timestamp(ColumnElement): type = TIMESTAMP() * :class:`~sqlalchemy.sql.functions.FunctionElement` - This is a hybrid of a ``ColumnElement`` and a "from clause" like object, and represents a SQL function or stored procedure type of call. Since most databases support statements along the line of "SELECT FROM <some function>" ``FunctionElement`` adds in the ability to be used in the FROM clause of a ``select()`` construct:: from sqlalchemy.sql.expression import FunctionElement class coalesce(FunctionElement): name = 'coalesce' @compiles(coalesce) def compile(element, compiler, kw): return "coalesce(%s)" % compiler.process(element.clauses, kw) @compiles(coalesce, 'oracle') def compile(element, compiler, kw): if len(element.clauses) > 2: raise TypeError("coalesce only supports two arguments on Oracle") return "nvl(%s)" % compiler.process(element.clauses, kw) * :class:`~sqlalchemy.schema.DDLElement` - The root of all DDL expressions, like CREATE TABLE, ALTER TABLE, etc. Compilation of ``DDLElement`` subclasses is issued by a ``DDLCompiler`` instead of a ``SQLCompiler``. ``DDLElement`` also features ``Table`` and ``MetaData`` event hooks via the ``execute_at()`` method, allowing the construct to be invoked during CREATE TABLE and DROP TABLE sequences. * :class:`~sqlalchemy.sql.expression.Executable` - This is a mixin which should be used with any expression class that represents a "standalone" SQL statement that can be passed directly to an ``execute()`` method. It is already implicit within ``DDLElement`` and ``FunctionElement``. Further Examples ================ "UTC timestamp" function ------------------------- A function that works like "CURRENT_TIMESTAMP" except applies the appropriate conversions so that the time is in UTC time. Timestamps are best stored in relational databases as UTC, without time zones. UTC so that your database doesn't think time has gone backwards in the hour when daylight savings ends, without timezones because timezones are like character encodings - they're best applied only at the endpoints of an application (i.e. convert to UTC upon user input, re-apply desired timezone upon display). For PostgreSQL and Microsoft SQL Server:: from sqlalchemy.sql import expression from sqlalchemy.ext.compiler import compiles from sqlalchemy.types import DateTime class utcnow(expression.FunctionElement): type = DateTime() @compiles(utcnow, 'postgresql') def pg_utcnow(element, compiler, kw): return "TIMEZONE('utc', CURRENT_TIMESTAMP)" @compiles(utcnow, 'mssql') def ms_utcnow(element, compiler, kw): return "GETUTCDATE()" Example usage:: from sqlalchemy import ( Table, Column, Integer, String, DateTime, MetaData ) metadata = MetaData() event = Table("event", metadata, Column("id", Integer, primary_key=True), Column("description", String(50), nullable=False), Column("timestamp", DateTime, server_default=utcnow()) ) "GREATEST" function ------------------- The "GREATEST" function is given any number of arguments and returns the one that is of the highest value - its equivalent to Python's ``max`` function. A SQL standard version versus a CASE based version which only accommodates two arguments:: from sqlalchemy.sql import expression, case from sqlalchemy.ext.compiler import compiles from sqlalchemy.types import Numeric class greatest(expression.FunctionElement): type = Numeric() name = 'greatest' @compiles(greatest) def default_greatest(element, compiler, kw): return compiler.visit_function(element) @compiles(greatest, 'sqlite') @compiles(greatest, 'mssql') @compiles(greatest, 'oracle') def case_greatest(element, compiler, kw): arg1, arg2 = list(element.clauses) return compiler.process(case([(arg1 > arg2, arg1)], else_=arg2), kw) Example usage:: Session.query(Account).\ filter( greatest( Account.checking_balance, Account.savings_balance) > 10000 ) "false" expression ------------------ Render a "false" constant expression, rendering as "0" on platforms that don't have a "false" constant:: from sqlalchemy.sql import expression from sqlalchemy.ext.compiler import compiles class sql_false(expression.ColumnElement): pass @compiles(sql_false) def default_false(element, compiler, kw): return "false" @compiles(sql_false, 'mssql') @compiles(sql_false, 'mysql') @compiles(sql_false, 'oracle') def int_false(element, compiler, *kw): return "0" Example usage:: from sqlalchemy import select, union_all exp = union_all( select([users.c.name, sql_false().label("enrolled")]), select([customers.c.name, customers.c.enrolled]) ) """ from .. import exc from .. import util from ..sql import visitors def compiles(class_, specs): """Register a function as a compiler for a given :class:`_expression.ClauseElement` type.""" def decorate(fn): # get an existing @compiles handler existing = class_.__dict__.get("_compiler_dispatcher", None) # get the original handler. All ClauseElement classes have one # of these, but some TypeEngine classes will not. existing_dispatch = getattr(class_, "_compiler_dispatch", None) if not existing: existing = _dispatcher() if existing_dispatch: def _wrap_existing_dispatch(element, compiler, kw): try: return existing_dispatch(element, compiler, kw) except exc.UnsupportedCompilationError as uce: util.raise_( exc.UnsupportedCompilationError( compiler, type(element), message="%s construct has no default " "compilation handler." % type(element), ), from_=uce, ) existing.specs["default"] = _wrap_existing_dispatch # TODO: why is the lambda needed ? setattr( class_, "_compiler_dispatch", lambda arg, kw: existing(arg, kw), ) setattr(class_, "_compiler_dispatcher", existing) if specs: for s in specs: existing.specs[s] = fn else: existing.specs["default"] = fn return fn return decorate def deregister(class_): """Remove all custom compilers associated with a given :class:`_expression.ClauseElement` type.""" if hasattr(class_, "_compiler_dispatcher"): # regenerate default _compiler_dispatch visitors._generate_dispatch(class_) # remove custom directive del class_._compiler_dispatcher class _dispatcher(object): def __init__(self): self.specs = {} def __call__(self, element, compiler, kw): # TODO: yes, this could also switch off of DBAPI in use. fn = self.specs.get(compiler.dialect.name, None) if not fn: try: fn = self.specs["default"] except KeyError as ke: util.raise_( exc.UnsupportedCompilationError( compiler, type(element), message="%s construct has no default " "compilation handler." % type(element), ), replace_context=ke, ) return fn(element, compiler, **kw)	16.95 KB	2026-01-20 12:58
📁 declarative	-	2026-03-21 15:45
📄 horizontal_shard.py # ext/horizontal_shard.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Horizontal sharding support. Defines a rudimental 'horizontal sharding' system which allows a Session to distribute queries and persistence operations across multiple databases. For a usage example, see the :ref:`examples_sharding` example included in the source distribution. """ from .. import inspect from .. import util from ..orm.query import Query from ..orm.session import Session __all__ = ["ShardedSession", "ShardedQuery"] class ShardedQuery(Query): def __init__(self, args, kwargs): super(ShardedQuery, self).__init__(args, kwargs) self.id_chooser = self.session.id_chooser self.query_chooser = self.session.query_chooser self._shard_id = None def set_shard(self, shard_id): """Return a new query, limited to a single shard ID. All subsequent operations with the returned query will be against the single shard regardless of other state. """ q = self._clone() q._shard_id = shard_id return q def _execute_and_instances(self, context): def iter_for_shard(shard_id): context.attributes["shard_id"] = context.identity_token = shard_id result = self._connection_from_session( mapper=self._bind_mapper(), shard_id=shard_id ).execute(context.statement, self._params) return self.instances(result, context) if context.identity_token is not None: return iter_for_shard(context.identity_token) elif self._shard_id is not None: return iter_for_shard(self._shard_id) else: partial = [] for shard_id in self.query_chooser(self): partial.extend(iter_for_shard(shard_id)) # if some kind of in memory 'sorting' # were done, this is where it would happen return iter(partial) def _execute_crud(self, stmt, mapper): def exec_for_shard(shard_id): conn = self._connection_from_session( mapper=mapper, shard_id=shard_id, clause=stmt, close_with_result=True, ) result = conn.execute(stmt, self._params) return result if self._shard_id is not None: return exec_for_shard(self._shard_id) else: rowcount = 0 results = [] for shard_id in self.query_chooser(self): result = exec_for_shard(shard_id) rowcount += result.rowcount results.append(result) return ShardedResult(results, rowcount) def _identity_lookup( self, mapper, primary_key_identity, identity_token=None, lazy_loaded_from=None, kw ): """Override the default Query._identity_lookup method so that we search for a given non-token primary key identity across all possible identity tokens (e.g. shard ids). """ if identity_token is not None: return super(ShardedQuery, self)._identity_lookup( mapper, primary_key_identity, identity_token=identity_token, kw ) else: q = self.session.query(mapper) if lazy_loaded_from: q = q._set_lazyload_from(lazy_loaded_from) for shard_id in self.id_chooser(q, primary_key_identity): obj = super(ShardedQuery, self)._identity_lookup( mapper, primary_key_identity, identity_token=shard_id, kw ) if obj is not None: return obj return None def _get_impl(self, primary_key_identity, db_load_fn, identity_token=None): """Override the default Query._get_impl() method so that we emit a query to the DB for each possible identity token, if we don't have one already. """ def _db_load_fn(query, primary_key_identity): # load from the database. The original db_load_fn will # use the given Query object to load from the DB, so our # shard_id is what will indicate the DB that we query from. if self._shard_id is not None: return db_load_fn(self, primary_key_identity) else: ident = util.to_list(primary_key_identity) # build a ShardedQuery for each shard identifier and # try to load from the DB for shard_id in self.id_chooser(self, ident): q = self.set_shard(shard_id) o = db_load_fn(q, ident) if o is not None: return o else: return None if identity_token is None and self._shard_id is not None: identity_token = self._shard_id return super(ShardedQuery, self)._get_impl( primary_key_identity, _db_load_fn, identity_token=identity_token ) class ShardedResult(object): """A value object that represents multiple :class:`_engine.ResultProxy` objects. This is used by the :meth:`.ShardedQuery._execute_crud` hook to return an object that takes the place of the single :class:`_engine.ResultProxy`. Attribute include ``result_proxies``, which is a sequence of the actual :class:`_engine.ResultProxy` objects, as well as ``aggregate_rowcount`` or ``rowcount``, which is the sum of all the individual rowcount values. .. versionadded:: 1.3 """ __slots__ = ("result_proxies", "aggregate_rowcount") def __init__(self, result_proxies, aggregate_rowcount): self.result_proxies = result_proxies self.aggregate_rowcount = aggregate_rowcount @property def rowcount(self): return self.aggregate_rowcount class ShardedSession(Session): def __init__( self, shard_chooser, id_chooser, query_chooser, shards=None, query_cls=ShardedQuery, kwargs ): """Construct a ShardedSession. :param shard_chooser: A callable which, passed a Mapper, a mapped instance, and possibly a SQL clause, returns a shard ID. This id may be based off of the attributes present within the object, or on some round-robin scheme. If the scheme is based on a selection, it should set whatever state on the instance to mark it in the future as participating in that shard. :param id_chooser: A callable, passed a query and a tuple of identity values, which should return a list of shard ids where the ID might reside. The databases will be queried in the order of this listing. :param query_chooser: For a given Query, returns the list of shard_ids where the query should be issued. Results from all shards returned will be combined together into a single listing. :param shards: A dictionary of string shard names to :class:`~sqlalchemy.engine.Engine` objects. """ super(ShardedSession, self).__init__(query_cls=query_cls, kwargs) self.shard_chooser = shard_chooser self.id_chooser = id_chooser self.query_chooser = query_chooser self.__binds = {} self.connection_callable = self.connection if shards is not None: for k in shards: self.bind_shard(k, shards[k]) def _choose_shard_and_assign(self, mapper, instance, kw): if instance is not None: state = inspect(instance) if state.key: token = state.key[2] assert token is not None return token elif state.identity_token: return state.identity_token shard_id = self.shard_chooser(mapper, instance, kw) if instance is not None: state.identity_token = shard_id return shard_id def connection(self, mapper=None, instance=None, shard_id=None, kwargs): if shard_id is None: shard_id = self._choose_shard_and_assign(mapper, instance) if self.transaction is not None: return self.transaction.connection(mapper, shard_id=shard_id) else: return self.get_bind( mapper, shard_id=shard_id, instance=instance )._contextual_connect(kwargs) def get_bind( self, mapper, shard_id=None, instance=None, clause=None, **kw ): if shard_id is None: shard_id = self._choose_shard_and_assign( mapper, instance, clause=clause ) return self.__binds[shard_id] def bind_shard(self, shard_id, bind): self.__binds[shard_id] = bind	8.92 KB	2026-01-20 12:58
📄 hybrid.py # ext/hybrid.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php r"""Define attributes on ORM-mapped classes that have "hybrid" behavior. "hybrid" means the attribute has distinct behaviors defined at the class level and at the instance level. The :mod:`~sqlalchemy.ext.hybrid` extension provides a special form of method decorator, is around 50 lines of code and has almost no dependencies on the rest of SQLAlchemy. It can, in theory, work with any descriptor-based expression system. Consider a mapping ``Interval``, representing integer ``start`` and ``end`` values. We can define higher level functions on mapped classes that produce SQL expressions at the class level, and Python expression evaluation at the instance level. Below, each function decorated with :class:`.hybrid_method` or :class:`.hybrid_property` may receive ``self`` as an instance of the class, or as the class itself:: from sqlalchemy import Column, Integer from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import Session, aliased from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method Base = declarative_base() class Interval(Base): __tablename__ = 'interval' id = Column(Integer, primary_key=True) start = Column(Integer, nullable=False) end = Column(Integer, nullable=False) def __init__(self, start, end): self.start = start self.end = end @hybrid_property def length(self): return self.end - self.start @hybrid_method def contains(self, point): return (self.start <= point) & (point <= self.end) @hybrid_method def intersects(self, other): return self.contains(other.start) \| self.contains(other.end) Above, the ``length`` property returns the difference between the ``end`` and ``start`` attributes. With an instance of ``Interval``, this subtraction occurs in Python, using normal Python descriptor mechanics:: >>> i1 = Interval(5, 10) >>> i1.length 5 When dealing with the ``Interval`` class itself, the :class:`.hybrid_property` descriptor evaluates the function body given the ``Interval`` class as the argument, which when evaluated with SQLAlchemy expression mechanics (here using the :attr:`.QueryableAttribute.expression` accessor) returns a new SQL expression:: >>> print(Interval.length.expression) interval."end" - interval.start >>> print(Session().query(Interval).filter(Interval.length > 10)) SELECT interval.id AS interval_id, interval.start AS interval_start, interval."end" AS interval_end FROM interval WHERE interval."end" - interval.start > :param_1 ORM methods such as :meth:`_query.Query.filter_by` generally use ``getattr()`` to locate attributes, so can also be used with hybrid attributes:: >>> print(Session().query(Interval).filter_by(length=5)) SELECT interval.id AS interval_id, interval.start AS interval_start, interval."end" AS interval_end FROM interval WHERE interval."end" - interval.start = :param_1 The ``Interval`` class example also illustrates two methods, ``contains()`` and ``intersects()``, decorated with :class:`.hybrid_method`. This decorator applies the same idea to methods that :class:`.hybrid_property` applies to attributes. The methods return boolean values, and take advantage of the Python ``\|`` and ``&`` bitwise operators to produce equivalent instance-level and SQL expression-level boolean behavior:: >>> i1.contains(6) True >>> i1.contains(15) False >>> i1.intersects(Interval(7, 18)) True >>> i1.intersects(Interval(25, 29)) False >>> print(Session().query(Interval).filter(Interval.contains(15))) SELECT interval.id AS interval_id, interval.start AS interval_start, interval."end" AS interval_end FROM interval WHERE interval.start <= :start_1 AND interval."end" > :end_1 >>> ia = aliased(Interval) >>> print(Session().query(Interval, ia).filter(Interval.intersects(ia))) SELECT interval.id AS interval_id, interval.start AS interval_start, interval."end" AS interval_end, interval_1.id AS interval_1_id, interval_1.start AS interval_1_start, interval_1."end" AS interval_1_end FROM interval, interval AS interval_1 WHERE interval.start <= interval_1.start AND interval."end" > interval_1.start OR interval.start <= interval_1."end" AND interval."end" > interval_1."end" .. _hybrid_distinct_expression: Defining Expression Behavior Distinct from Attribute Behavior -------------------------------------------------------------- Our usage of the ``&`` and ``\|`` bitwise operators above was fortunate, considering our functions operated on two boolean values to return a new one. In many cases, the construction of an in-Python function and a SQLAlchemy SQL expression have enough differences that two separate Python expressions should be defined. The :mod:`~sqlalchemy.ext.hybrid` decorators define the :meth:`.hybrid_property.expression` modifier for this purpose. As an example we'll define the radius of the interval, which requires the usage of the absolute value function:: from sqlalchemy import func class Interval(object): # ... @hybrid_property def radius(self): return abs(self.length) / 2 @radius.expression def radius(cls): return func.abs(cls.length) / 2 Above the Python function ``abs()`` is used for instance-level operations, the SQL function ``ABS()`` is used via the :data:`.func` object for class-level expressions:: >>> i1.radius 2 >>> print(Session().query(Interval).filter(Interval.radius > 5)) SELECT interval.id AS interval_id, interval.start AS interval_start, interval."end" AS interval_end FROM interval WHERE abs(interval."end" - interval.start) / :abs_1 > :param_1 .. note:: When defining an expression for a hybrid property or method, the expression method must retain the name of the original hybrid, else the new hybrid with the additional state will be attached to the class with the non-matching name. To use the example above:: class Interval(object): # ... @hybrid_property def radius(self): return abs(self.length) / 2 # WRONG - the non-matching name will cause this function to be # ignored @radius.expression def radius_expression(cls): return func.abs(cls.length) / 2 This is also true for other mutator methods, such as :meth:`.hybrid_property.update_expression`. This is the same behavior as that of the ``@property`` construct that is part of standard Python. Defining Setters ---------------- Hybrid properties can also define setter methods. If we wanted ``length`` above, when set, to modify the endpoint value:: class Interval(object): # ... @hybrid_property def length(self): return self.end - self.start @length.setter def length(self, value): self.end = self.start + value The ``length(self, value)`` method is now called upon set:: >>> i1 = Interval(5, 10) >>> i1.length 5 >>> i1.length = 12 >>> i1.end 17 .. _hybrid_bulk_update: Allowing Bulk ORM Update ------------------------ A hybrid can define a custom "UPDATE" handler for when using the :meth:`_query.Query.update` method, allowing the hybrid to be used in the SET clause of the update. Normally, when using a hybrid with :meth:`_query.Query.update`, the SQL expression is used as the column that's the target of the SET. If our ``Interval`` class had a hybrid ``start_point`` that linked to ``Interval.start``, this could be substituted directly:: session.query(Interval).update({Interval.start_point: 10}) However, when using a composite hybrid like ``Interval.length``, this hybrid represents more than one column. We can set up a handler that will accommodate a value passed to :meth:`_query.Query.update` which can affect this, using the :meth:`.hybrid_property.update_expression` decorator. A handler that works similarly to our setter would be:: class Interval(object): # ... @hybrid_property def length(self): return self.end - self.start @length.setter def length(self, value): self.end = self.start + value @length.update_expression def length(cls, value): return [ (cls.end, cls.start + value) ] Above, if we use ``Interval.length`` in an UPDATE expression as:: session.query(Interval).update( {Interval.length: 25}, synchronize_session='fetch') We'll get an UPDATE statement along the lines of:: UPDATE interval SET end=start + :value In some cases, the default "evaluate" strategy can't perform the SET expression in Python; while the addition operator we're using above is supported, for more complex SET expressions it will usually be necessary to use either the "fetch" or False synchronization strategy as illustrated above. .. versionadded:: 1.2 added support for bulk updates to hybrid properties. Working with Relationships -------------------------- There's no essential difference when creating hybrids that work with related objects as opposed to column-based data. The need for distinct expressions tends to be greater. The two variants we'll illustrate are the "join-dependent" hybrid, and the "correlated subquery" hybrid. Join-Dependent Relationship Hybrid ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ Consider the following declarative mapping which relates a ``User`` to a ``SavingsAccount``:: from sqlalchemy import Column, Integer, ForeignKey, Numeric, String from sqlalchemy.orm import relationship from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.ext.hybrid import hybrid_property Base = declarative_base() class SavingsAccount(Base): __tablename__ = 'account' id = Column(Integer, primary_key=True) user_id = Column(Integer, ForeignKey('user.id'), nullable=False) balance = Column(Numeric(15, 5)) class User(Base): __tablename__ = 'user' id = Column(Integer, primary_key=True) name = Column(String(100), nullable=False) accounts = relationship("SavingsAccount", backref="owner") @hybrid_property def balance(self): if self.accounts: return self.accounts[0].balance else: return None @balance.setter def balance(self, value): if not self.accounts: account = Account(owner=self) else: account = self.accounts[0] account.balance = value @balance.expression def balance(cls): return SavingsAccount.balance The above hybrid property ``balance`` works with the first ``SavingsAccount`` entry in the list of accounts for this user. The in-Python getter/setter methods can treat ``accounts`` as a Python list available on ``self``. However, at the expression level, it's expected that the ``User`` class will be used in an appropriate context such that an appropriate join to ``SavingsAccount`` will be present:: >>> print(Session().query(User, User.balance). ... join(User.accounts).filter(User.balance > 5000)) SELECT "user".id AS user_id, "user".name AS user_name, account.balance AS account_balance FROM "user" JOIN account ON "user".id = account.user_id WHERE account.balance > :balance_1 Note however, that while the instance level accessors need to worry about whether ``self.accounts`` is even present, this issue expresses itself differently at the SQL expression level, where we basically would use an outer join:: >>> from sqlalchemy import or_ >>> print (Session().query(User, User.balance).outerjoin(User.accounts). ... filter(or_(User.balance < 5000, User.balance == None))) SELECT "user".id AS user_id, "user".name AS user_name, account.balance AS account_balance FROM "user" LEFT OUTER JOIN account ON "user".id = account.user_id WHERE account.balance < :balance_1 OR account.balance IS NULL Correlated Subquery Relationship Hybrid ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ We can, of course, forego being dependent on the enclosing query's usage of joins in favor of the correlated subquery, which can portably be packed into a single column expression. A correlated subquery is more portable, but often performs more poorly at the SQL level. Using the same technique illustrated at :ref:`mapper_column_property_sql_expressions`, we can adjust our ``SavingsAccount`` example to aggregate the balances for all accounts, and use a correlated subquery for the column expression:: from sqlalchemy import Column, Integer, ForeignKey, Numeric, String from sqlalchemy.orm import relationship from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.ext.hybrid import hybrid_property from sqlalchemy import select, func Base = declarative_base() class SavingsAccount(Base): __tablename__ = 'account' id = Column(Integer, primary_key=True) user_id = Column(Integer, ForeignKey('user.id'), nullable=False) balance = Column(Numeric(15, 5)) class User(Base): __tablename__ = 'user' id = Column(Integer, primary_key=True) name = Column(String(100), nullable=False) accounts = relationship("SavingsAccount", backref="owner") @hybrid_property def balance(self): return sum(acc.balance for acc in self.accounts) @balance.expression def balance(cls): return select([func.sum(SavingsAccount.balance)]).\ where(SavingsAccount.user_id==cls.id).\ label('total_balance') The above recipe will give us the ``balance`` column which renders a correlated SELECT:: >>> print(s.query(User).filter(User.balance > 400)) SELECT "user".id AS user_id, "user".name AS user_name FROM "user" WHERE (SELECT sum(account.balance) AS sum_1 FROM account WHERE account.user_id = "user".id) > :param_1 .. _hybrid_custom_comparators: Building Custom Comparators --------------------------- The hybrid property also includes a helper that allows construction of custom comparators. A comparator object allows one to customize the behavior of each SQLAlchemy expression operator individually. They are useful when creating custom types that have some highly idiosyncratic behavior on the SQL side. .. note:: The :meth:`.hybrid_property.comparator` decorator introduced in this section replaces the use of the :meth:`.hybrid_property.expression` decorator. They cannot be used together. The example class below allows case-insensitive comparisons on the attribute named ``word_insensitive``:: from sqlalchemy.ext.hybrid import Comparator, hybrid_property from sqlalchemy import func, Column, Integer, String from sqlalchemy.orm import Session from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class CaseInsensitiveComparator(Comparator): def __eq__(self, other): return func.lower(self.__clause_element__()) == func.lower(other) class SearchWord(Base): __tablename__ = 'searchword' id = Column(Integer, primary_key=True) word = Column(String(255), nullable=False) @hybrid_property def word_insensitive(self): return self.word.lower() @word_insensitive.comparator def word_insensitive(cls): return CaseInsensitiveComparator(cls.word) Above, SQL expressions against ``word_insensitive`` will apply the ``LOWER()`` SQL function to both sides:: >>> print(Session().query(SearchWord).filter_by(word_insensitive="Trucks")) SELECT searchword.id AS searchword_id, searchword.word AS searchword_word FROM searchword WHERE lower(searchword.word) = lower(:lower_1) The ``CaseInsensitiveComparator`` above implements part of the :class:`.ColumnOperators` interface. A "coercion" operation like lowercasing can be applied to all comparison operations (i.e. ``eq``, ``lt``, ``gt``, etc.) using :meth:`.Operators.operate`:: class CaseInsensitiveComparator(Comparator): def operate(self, op, other): return op(func.lower(self.__clause_element__()), func.lower(other)) .. _hybrid_reuse_subclass: Reusing Hybrid Properties across Subclasses ------------------------------------------- A hybrid can be referred to from a superclass, to allow modifying methods like :meth:`.hybrid_property.getter`, :meth:`.hybrid_property.setter` to be used to redefine those methods on a subclass. This is similar to how the standard Python ``@property`` object works:: class FirstNameOnly(Base): # ... first_name = Column(String) @hybrid_property def name(self): return self.first_name @name.setter def name(self, value): self.first_name = value class FirstNameLastName(FirstNameOnly): # ... last_name = Column(String) @FirstNameOnly.name.getter def name(self): return self.first_name + ' ' + self.last_name @name.setter def name(self, value): self.first_name, self.last_name = value.split(' ', 1) Above, the ``FirstNameLastName`` class refers to the hybrid from ``FirstNameOnly.name`` to repurpose its getter and setter for the subclass. When overriding :meth:`.hybrid_property.expression` and :meth:`.hybrid_property.comparator` alone as the first reference to the superclass, these names conflict with the same-named accessors on the class- level :class:`.QueryableAttribute` object returned at the class level. To override these methods when referring directly to the parent class descriptor, add the special qualifier :attr:`.hybrid_property.overrides`, which will de- reference the instrumented attribute back to the hybrid object:: class FirstNameLastName(FirstNameOnly): # ... last_name = Column(String) @FirstNameOnly.name.overrides.expression def name(cls): return func.concat(cls.first_name, ' ', cls.last_name) .. versionadded:: 1.2 Added :meth:`.hybrid_property.getter` as well as the ability to redefine accessors per-subclass. Hybrid Value Objects -------------------- Note in our previous example, if we were to compare the ``word_insensitive`` attribute of a ``SearchWord`` instance to a plain Python string, the plain Python string would not be coerced to lower case - the ``CaseInsensitiveComparator`` we built, being returned by ``@word_insensitive.comparator``, only applies to the SQL side. A more comprehensive form of the custom comparator is to construct a Hybrid Value Object. This technique applies the target value or expression to a value object which is then returned by the accessor in all cases. The value object allows control of all operations upon the value as well as how compared values are treated, both on the SQL expression side as well as the Python value side. Replacing the previous ``CaseInsensitiveComparator`` class with a new ``CaseInsensitiveWord`` class:: class CaseInsensitiveWord(Comparator): "Hybrid value representing a lower case representation of a word." def __init__(self, word): if isinstance(word, basestring): self.word = word.lower() elif isinstance(word, CaseInsensitiveWord): self.word = word.word else: self.word = func.lower(word) def operate(self, op, other): if not isinstance(other, CaseInsensitiveWord): other = CaseInsensitiveWord(other) return op(self.word, other.word) def __clause_element__(self): return self.word def __str__(self): return self.word key = 'word' "Label to apply to Query tuple results" Above, the ``CaseInsensitiveWord`` object represents ``self.word``, which may be a SQL function, or may be a Python native. By overriding ``operate()`` and ``__clause_element__()`` to work in terms of ``self.word``, all comparison operations will work against the "converted" form of ``word``, whether it be SQL side or Python side. Our ``SearchWord`` class can now deliver the ``CaseInsensitiveWord`` object unconditionally from a single hybrid call:: class SearchWord(Base): __tablename__ = 'searchword' id = Column(Integer, primary_key=True) word = Column(String(255), nullable=False) @hybrid_property def word_insensitive(self): return CaseInsensitiveWord(self.word) The ``word_insensitive`` attribute now has case-insensitive comparison behavior universally, including SQL expression vs. Python expression (note the Python value is converted to lower case on the Python side here):: >>> print(Session().query(SearchWord).filter_by(word_insensitive="Trucks")) SELECT searchword.id AS searchword_id, searchword.word AS searchword_word FROM searchword WHERE lower(searchword.word) = :lower_1 SQL expression versus SQL expression:: >>> sw1 = aliased(SearchWord) >>> sw2 = aliased(SearchWord) >>> print(Session().query( ... sw1.word_insensitive, ... sw2.word_insensitive).\ ... filter( ... sw1.word_insensitive > sw2.word_insensitive ... )) SELECT lower(searchword_1.word) AS lower_1, lower(searchword_2.word) AS lower_2 FROM searchword AS searchword_1, searchword AS searchword_2 WHERE lower(searchword_1.word) > lower(searchword_2.word) Python only expression:: >>> ws1 = SearchWord(word="SomeWord") >>> ws1.word_insensitive == "sOmEwOrD" True >>> ws1.word_insensitive == "XOmEwOrX" False >>> print(ws1.word_insensitive) someword The Hybrid Value pattern is very useful for any kind of value that may have multiple representations, such as timestamps, time deltas, units of measurement, currencies and encrypted passwords. .. seealso:: `Hybrids and Value Agnostic Types <http://techspot.zzzeek.org/2011/10/21/hybrids-and-value-agnostic-types/>`_ - on the techspot.zzzeek.org blog `Value Agnostic Types, Part II <http://techspot.zzzeek.org/2011/10/29/value-agnostic-types-part-ii/>`_ - on the techspot.zzzeek.org blog .. _hybrid_transformers: Building Transformers ---------------------- A transformer is an object which can receive a :class:`_query.Query` object and return a new one. The :class:`_query.Query` object includes a method :meth:`.with_transformation` that returns a new :class:`_query.Query` transformed by the given function. We can combine this with the :class:`.Comparator` class to produce one type of recipe which can both set up the FROM clause of a query as well as assign filtering criterion. Consider a mapped class ``Node``, which assembles using adjacency list into a hierarchical tree pattern:: from sqlalchemy import Column, Integer, ForeignKey from sqlalchemy.orm import relationship from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class Node(Base): __tablename__ = 'node' id = Column(Integer, primary_key=True) parent_id = Column(Integer, ForeignKey('node.id')) parent = relationship("Node", remote_side=id) Suppose we wanted to add an accessor ``grandparent``. This would return the ``parent`` of ``Node.parent``. When we have an instance of ``Node``, this is simple:: from sqlalchemy.ext.hybrid import hybrid_property class Node(Base): # ... @hybrid_property def grandparent(self): return self.parent.parent For the expression, things are not so clear. We'd need to construct a :class:`_query.Query` where we :meth:`_query.Query.join` twice along ``Node.parent`` to get to the ``grandparent``. We can instead return a transforming callable that we'll combine with the :class:`.Comparator` class to receive any :class:`_query.Query` object, and return a new one that's joined to the ``Node.parent`` attribute and filtered based on the given criterion:: from sqlalchemy.ext.hybrid import Comparator class GrandparentTransformer(Comparator): def operate(self, op, other): def transform(q): cls = self.__clause_element__() parent_alias = aliased(cls) return q.join(parent_alias, cls.parent).\ filter(op(parent_alias.parent, other)) return transform Base = declarative_base() class Node(Base): __tablename__ = 'node' id = Column(Integer, primary_key=True) parent_id = Column(Integer, ForeignKey('node.id')) parent = relationship("Node", remote_side=id) @hybrid_property def grandparent(self): return self.parent.parent @grandparent.comparator def grandparent(cls): return GrandparentTransformer(cls) The ``GrandparentTransformer`` overrides the core :meth:`.Operators.operate` method at the base of the :class:`.Comparator` hierarchy to return a query- transforming callable, which then runs the given comparison operation in a particular context. Such as, in the example above, the ``operate`` method is called, given the :attr:`.Operators.eq` callable as well as the right side of the comparison ``Node(id=5)``. A function ``transform`` is then returned which will transform a :class:`_query.Query` first to join to ``Node.parent``, then to compare ``parent_alias`` using :attr:`.Operators.eq` against the left and right sides, passing into :meth:`_query.Query.filter`: .. sourcecode:: pycon+sql >>> from sqlalchemy.orm import Session >>> session = Session() {sql}>>> session.query(Node).\ ... with_transformation(Node.grandparent==Node(id=5)).\ ... all() SELECT node.id AS node_id, node.parent_id AS node_parent_id FROM node JOIN node AS node_1 ON node_1.id = node.parent_id WHERE :param_1 = node_1.parent_id {stop} We can modify the pattern to be more verbose but flexible by separating the "join" step from the "filter" step. The tricky part here is ensuring that successive instances of ``GrandparentTransformer`` use the same :class:`.AliasedClass` object against ``Node``. Below we use a simple memoizing approach that associates a ``GrandparentTransformer`` with each class:: class Node(Base): # ... @grandparent.comparator def grandparent(cls): # memoize a GrandparentTransformer # per class if '_gp' not in cls.__dict__: cls._gp = GrandparentTransformer(cls) return cls._gp class GrandparentTransformer(Comparator): def __init__(self, cls): self.parent_alias = aliased(cls) @property def join(self): def go(q): return q.join(self.parent_alias, Node.parent) return go def operate(self, op, other): return op(self.parent_alias.parent, other) .. sourcecode:: pycon+sql {sql}>>> session.query(Node).\ ... with_transformation(Node.grandparent.join).\ ... filter(Node.grandparent==Node(id=5)) SELECT node.id AS node_id, node.parent_id AS node_parent_id FROM node JOIN node AS node_1 ON node_1.id = node.parent_id WHERE :param_1 = node_1.parent_id {stop} The "transformer" pattern is an experimental pattern that starts to make usage of some functional programming paradigms. While it's only recommended for advanced and/or patient developers, there's probably a whole lot of amazing things it can be used for. """ # noqa from .. import util from ..orm import attributes from ..orm import interfaces HYBRID_METHOD = util.symbol("HYBRID_METHOD") """Symbol indicating an :class:`InspectionAttr` that's of type :class:`.hybrid_method`. Is assigned to the :attr:`.InspectionAttr.extension_type` attribute. .. seealso:: :attr:`_orm.Mapper.all_orm_attributes` """ HYBRID_PROPERTY = util.symbol("HYBRID_PROPERTY") """Symbol indicating an :class:`InspectionAttr` that's of type :class:`.hybrid_method`. Is assigned to the :attr:`.InspectionAttr.extension_type` attribute. .. seealso:: :attr:`_orm.Mapper.all_orm_attributes` """ class hybrid_method(interfaces.InspectionAttrInfo): """A decorator which allows definition of a Python object method with both instance-level and class-level behavior. """ is_attribute = True extension_type = HYBRID_METHOD def __init__(self, func, expr=None): """Create a new :class:`.hybrid_method`. Usage is typically via decorator:: from sqlalchemy.ext.hybrid import hybrid_method class SomeClass(object): @hybrid_method def value(self, x, y): return self._value + x + y @value.expression def value(self, x, y): return func.some_function(self._value, x, y) """ self.func = func self.expression(expr or func) def __get__(self, instance, owner): if instance is None: return self.expr.__get__(owner, owner.__class__) else: return self.func.__get__(instance, owner) def expression(self, expr): """Provide a modifying decorator that defines a SQL-expression producing method.""" self.expr = expr if not self.expr.__doc__: self.expr.__doc__ = self.func.__doc__ return self class hybrid_property(interfaces.InspectionAttrInfo): """A decorator which allows definition of a Python descriptor with both instance-level and class-level behavior. """ is_attribute = True extension_type = HYBRID_PROPERTY def __init__( self, fget, fset=None, fdel=None, expr=None, custom_comparator=None, update_expr=None, ): """Create a new :class:`.hybrid_property`. Usage is typically via decorator:: from sqlalchemy.ext.hybrid import hybrid_property class SomeClass(object): @hybrid_property def value(self): return self._value @value.setter def value(self, value): self._value = value """ self.fget = fget self.fset = fset self.fdel = fdel self.expr = expr self.custom_comparator = custom_comparator self.update_expr = update_expr util.update_wrapper(self, fget) def __get__(self, instance, owner): if instance is None: return self._expr_comparator(owner) else: return self.fget(instance) def __set__(self, instance, value): if self.fset is None: raise AttributeError("can't set attribute") self.fset(instance, value) def __delete__(self, instance): if self.fdel is None: raise AttributeError("can't delete attribute") self.fdel(instance) def _copy(self, kw): defaults = { key: value for key, value in self.__dict__.items() if not key.startswith("_") } defaults.update(kw) return type(self)(defaults) @property def overrides(self): """Prefix for a method that is overriding an existing attribute. The :attr:`.hybrid_property.overrides` accessor just returns this hybrid object, which when called at the class level from a parent class, will de-reference the "instrumented attribute" normally returned at this level, and allow modifying decorators like :meth:`.hybrid_property.expression` and :meth:`.hybrid_property.comparator` to be used without conflicting with the same-named attributes normally present on the :class:`.QueryableAttribute`:: class SuperClass(object): # ... @hybrid_property def foobar(self): return self._foobar class SubClass(SuperClass): # ... @SuperClass.foobar.overrides.expression def foobar(cls): return func.subfoobar(self._foobar) .. versionadded:: 1.2 .. seealso:: :ref:`hybrid_reuse_subclass` """ return self def getter(self, fget): """Provide a modifying decorator that defines a getter method. .. versionadded:: 1.2 """ return self._copy(fget=fget) def setter(self, fset): """Provide a modifying decorator that defines a setter method.""" return self._copy(fset=fset) def deleter(self, fdel): """Provide a modifying decorator that defines a deletion method.""" return self._copy(fdel=fdel) def expression(self, expr): """Provide a modifying decorator that defines a SQL-expression producing method. When a hybrid is invoked at the class level, the SQL expression given here is wrapped inside of a specialized :class:`.QueryableAttribute`, which is the same kind of object used by the ORM to represent other mapped attributes. The reason for this is so that other class-level attributes such as docstrings and a reference to the hybrid itself may be maintained within the structure that's returned, without any modifications to the original SQL expression passed in. .. note:: When referring to a hybrid property from an owning class (e.g. ``SomeClass.some_hybrid``), an instance of :class:`.QueryableAttribute` is returned, representing the expression or comparator object as well as this hybrid object. However, that object itself has accessors called ``expression`` and ``comparator``; so when attempting to override these decorators on a subclass, it may be necessary to qualify it using the :attr:`.hybrid_property.overrides` modifier first. See that modifier for details. .. seealso:: :ref:`hybrid_distinct_expression` """ return self._copy(expr=expr) def comparator(self, comparator): """Provide a modifying decorator that defines a custom comparator producing method. The return value of the decorated method should be an instance of :class:`~.hybrid.Comparator`. .. note:: The :meth:`.hybrid_property.comparator` decorator replaces** the use of the :meth:`.hybrid_property.expression` decorator. They cannot be used together. When a hybrid is invoked at the class level, the :class:`~.hybrid.Comparator` object given here is wrapped inside of a specialized :class:`.QueryableAttribute`, which is the same kind of object used by the ORM to represent other mapped attributes. The reason for this is so that other class-level attributes such as docstrings and a reference to the hybrid itself may be maintained within the structure that's returned, without any modifications to the original comparator object passed in. .. note:: When referring to a hybrid property from an owning class (e.g. ``SomeClass.some_hybrid``), an instance of :class:`.QueryableAttribute` is returned, representing the expression or comparator object as this hybrid object. However, that object itself has accessors called ``expression`` and ``comparator``; so when attempting to override these decorators on a subclass, it may be necessary to qualify it using the :attr:`.hybrid_property.overrides` modifier first. See that modifier for details. """ return self._copy(custom_comparator=comparator) def update_expression(self, meth): """Provide a modifying decorator that defines an UPDATE tuple producing method. The method accepts a single value, which is the value to be rendered into the SET clause of an UPDATE statement. The method should then process this value into individual column expressions that fit into the ultimate SET clause, and return them as a sequence of 2-tuples. Each tuple contains a column expression as the key and a value to be rendered. E.g.:: class Person(Base): # ... first_name = Column(String) last_name = Column(String) @hybrid_property def fullname(self): return first_name + " " + last_name @fullname.update_expression def fullname(cls, value): fname, lname = value.split(" ", 1) return [ (cls.first_name, fname), (cls.last_name, lname) ] .. versionadded:: 1.2 """ return self._copy(update_expr=meth) @util.memoized_property def _expr_comparator(self): if self.custom_comparator is not None: return self._get_comparator(self.custom_comparator) elif self.expr is not None: return self._get_expr(self.expr) else: return self._get_expr(self.fget) def _get_expr(self, expr): def _expr(cls): return ExprComparator(cls, expr(cls), self) util.update_wrapper(_expr, expr) return self._get_comparator(_expr) def _get_comparator(self, comparator): proxy_attr = attributes.create_proxied_attribute(self) def expr_comparator(owner): return proxy_attr( owner, self.__name__, self, comparator(owner), doc=comparator.__doc__ or self.__doc__, ) return expr_comparator class Comparator(interfaces.PropComparator): """A helper class that allows easy construction of custom :class:`~.orm.interfaces.PropComparator` classes for usage with hybrids.""" property = None def __init__(self, expression): self.expression = expression def __clause_element__(self): expr = self.expression if hasattr(expr, "__clause_element__"): expr = expr.__clause_element__() return expr def adapt_to_entity(self, adapt_to_entity): # interesting.... return self class ExprComparator(Comparator): def __init__(self, cls, expression, hybrid): self.cls = cls self.expression = expression self.hybrid = hybrid def __getattr__(self, key): return getattr(self.expression, key) @property def info(self): return self.hybrid.info def _bulk_update_tuples(self, value): if isinstance(self.expression, attributes.QueryableAttribute): return self.expression._bulk_update_tuples(value) elif self.hybrid.update_expr is not None: return self.hybrid.update_expr(self.cls, value) else: return [(self.expression, value)] @property def property(self): return self.expression.property def operate(self, op, other, kwargs): return op(self.expression, other, kwargs) def reverse_operate(self, op, other, kwargs): return op(other, self.expression, **kwargs)	39.45 KB	2026-01-20 12:58
📄 indexable.py # ext/index.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Define attributes on ORM-mapped classes that have "index" attributes for columns with :class:`_types.Indexable` types. "index" means the attribute is associated with an element of an :class:`_types.Indexable` column with the predefined index to access it. The :class:`_types.Indexable` types include types such as :class:`_types.ARRAY`, :class:`_types.JSON` and :class:`_postgresql.HSTORE`. The :mod:`~sqlalchemy.ext.indexable` extension provides :class:`_schema.Column`-like interface for any element of an :class:`_types.Indexable` typed column. In simple cases, it can be treated as a :class:`_schema.Column` - mapped attribute. .. versionadded:: 1.1 Synopsis ======== Given ``Person`` as a model with a primary key and JSON data field. While this field may have any number of elements encoded within it, we would like to refer to the element called ``name`` individually as a dedicated attribute which behaves like a standalone column:: from sqlalchemy import Column, JSON, Integer from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.ext.indexable import index_property Base = declarative_base() class Person(Base): __tablename__ = 'person' id = Column(Integer, primary_key=True) data = Column(JSON) name = index_property('data', 'name') Above, the ``name`` attribute now behaves like a mapped column. We can compose a new ``Person`` and set the value of ``name``:: >>> person = Person(name='Alchemist') The value is now accessible:: >>> person.name 'Alchemist' Behind the scenes, the JSON field was initialized to a new blank dictionary and the field was set:: >>> person.data {"name": "Alchemist'} The field is mutable in place:: >>> person.name = 'Renamed' >>> person.name 'Renamed' >>> person.data {'name': 'Renamed'} When using :class:`.index_property`, the change that we make to the indexable structure is also automatically tracked as history; we no longer need to use :class:`~.mutable.MutableDict` in order to track this change for the unit of work. Deletions work normally as well:: >>> del person.name >>> person.data {} Above, deletion of ``person.name`` deletes the value from the dictionary, but not the dictionary itself. A missing key will produce ``AttributeError``:: >>> person = Person() >>> person.name ... AttributeError: 'name' Unless you set a default value:: >>> class Person(Base): >>> __tablename__ = 'person' >>> >>> id = Column(Integer, primary_key=True) >>> data = Column(JSON) >>> >>> name = index_property('data', 'name', default=None) # See default >>> person = Person() >>> print(person.name) None The attributes are also accessible at the class level. Below, we illustrate ``Person.name`` used to generate an indexed SQL criteria:: >>> from sqlalchemy.orm import Session >>> session = Session() >>> query = session.query(Person).filter(Person.name == 'Alchemist') The above query is equivalent to:: >>> query = session.query(Person).filter(Person.data['name'] == 'Alchemist') Multiple :class:`.index_property` objects can be chained to produce multiple levels of indexing:: from sqlalchemy import Column, JSON, Integer from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.ext.indexable import index_property Base = declarative_base() class Person(Base): __tablename__ = 'person' id = Column(Integer, primary_key=True) data = Column(JSON) birthday = index_property('data', 'birthday') year = index_property('birthday', 'year') month = index_property('birthday', 'month') day = index_property('birthday', 'day') Above, a query such as:: q = session.query(Person).filter(Person.year == '1980') On a PostgreSQL backend, the above query will render as:: SELECT person.id, person.data FROM person WHERE person.data -> %(data_1)s -> %(param_1)s = %(param_2)s Default Values ============== :class:`.index_property` includes special behaviors for when the indexed data structure does not exist, and a set operation is called: * For an :class:`.index_property` that is given an integer index value, the default data structure will be a Python list of ``None`` values, at least as long as the index value; the value is then set at its place in the list. This means for an index value of zero, the list will be initialized to ``[None]`` before setting the given value, and for an index value of five, the list will be initialized to ``[None, None, None, None, None]`` before setting the fifth element to the given value. Note that an existing list is not extended in place to receive a value. * for an :class:`.index_property` that is given any other kind of index value (e.g. strings usually), a Python dictionary is used as the default data structure. * The default data structure can be set to any Python callable using the :paramref:`.index_property.datatype` parameter, overriding the previous rules. Subclassing =========== :class:`.index_property` can be subclassed, in particular for the common use case of providing coercion of values or SQL expressions as they are accessed. Below is a common recipe for use with a PostgreSQL JSON type, where we want to also include automatic casting plus ``astext()``:: class pg_json_property(index_property): def __init__(self, attr_name, index, cast_type): super(pg_json_property, self).__init__(attr_name, index) self.cast_type = cast_type def expr(self, model): expr = super(pg_json_property, self).expr(model) return expr.astext.cast(self.cast_type) The above subclass can be used with the PostgreSQL-specific version of :class:`_postgresql.JSON`:: from sqlalchemy import Column, Integer from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.dialects.postgresql import JSON Base = declarative_base() class Person(Base): __tablename__ = 'person' id = Column(Integer, primary_key=True) data = Column(JSON) age = pg_json_property('data', 'age', Integer) The ``age`` attribute at the instance level works as before; however when rendering SQL, PostgreSQL's ``->>`` operator will be used for indexed access, instead of the usual index operator of ``->``:: >>> query = session.query(Person).filter(Person.age < 20) The above query will render:: SELECT person.id, person.data FROM person WHERE CAST(person.data ->> %(data_1)s AS INTEGER) < %(param_1)s """ # noqa from __future__ import absolute_import from .. import inspect from .. import util from ..ext.hybrid import hybrid_property from ..orm.attributes import flag_modified __all__ = ["index_property"] class index_property(hybrid_property): # noqa """A property generator. The generated property describes an object attribute that corresponds to an :class:`_types.Indexable` column. .. versionadded:: 1.1 .. seealso:: :mod:`sqlalchemy.ext.indexable` """ _NO_DEFAULT_ARGUMENT = object() def __init__( self, attr_name, index, default=_NO_DEFAULT_ARGUMENT, datatype=None, mutable=True, onebased=True, ): """Create a new :class:`.index_property`. :param attr_name: An attribute name of an `Indexable` typed column, or other attribute that returns an indexable structure. :param index: The index to be used for getting and setting this value. This should be the Python-side index value for integers. :param default: A value which will be returned instead of `AttributeError` when there is not a value at given index. :param datatype: default datatype to use when the field is empty. By default, this is derived from the type of index used; a Python list for an integer index, or a Python dictionary for any other style of index. For a list, the list will be initialized to a list of None values that is at least ``index`` elements long. :param mutable: if False, writes and deletes to the attribute will be disallowed. :param onebased: assume the SQL representation of this value is one-based; that is, the first index in SQL is 1, not zero. """ if mutable: super(index_property, self).__init__( self.fget, self.fset, self.fdel, self.expr ) else: super(index_property, self).__init__( self.fget, None, None, self.expr ) self.attr_name = attr_name self.index = index self.default = default is_numeric = isinstance(index, int) onebased = is_numeric and onebased if datatype is not None: self.datatype = datatype else: if is_numeric: self.datatype = lambda: [None for x in range(index + 1)] else: self.datatype = dict self.onebased = onebased def _fget_default(self, err=None): if self.default == self._NO_DEFAULT_ARGUMENT: util.raise_(AttributeError(self.attr_name), replace_context=err) else: return self.default def fget(self, instance): attr_name = self.attr_name column_value = getattr(instance, attr_name) if column_value is None: return self._fget_default() try: value = column_value[self.index] except (KeyError, IndexError) as err: return self._fget_default(err) else: return value def fset(self, instance, value): attr_name = self.attr_name column_value = getattr(instance, attr_name, None) if column_value is None: column_value = self.datatype() setattr(instance, attr_name, column_value) column_value[self.index] = value setattr(instance, attr_name, column_value) if attr_name in inspect(instance).mapper.attrs: flag_modified(instance, attr_name) def fdel(self, instance): attr_name = self.attr_name column_value = getattr(instance, attr_name) if column_value is None: raise AttributeError(self.attr_name) try: del column_value[self.index] except KeyError as err: util.raise_(AttributeError(self.attr_name), replace_context=err) else: setattr(instance, attr_name, column_value) flag_modified(instance, attr_name) def expr(self, model): column = getattr(model, self.attr_name) index = self.index if self.onebased: index += 1 return column[index]	10.99 KB	2026-01-20 12:58
📄 instrumentation.py """Extensible class instrumentation. The :mod:`sqlalchemy.ext.instrumentation` package provides for alternate systems of class instrumentation within the ORM. Class instrumentation refers to how the ORM places attributes on the class which maintain data and track changes to that data, as well as event hooks installed on the class. .. note:: The extension package is provided for the benefit of integration with other object management packages, which already perform their own instrumentation. It is not intended for general use. For examples of how the instrumentation extension is used, see the example :ref:`examples_instrumentation`. """ import weakref from .. import util from ..orm import attributes from ..orm import base as orm_base from ..orm import collections from ..orm import exc as orm_exc from ..orm import instrumentation as orm_instrumentation from ..orm.instrumentation import _default_dict_getter from ..orm.instrumentation import _default_manager_getter from ..orm.instrumentation import _default_state_getter from ..orm.instrumentation import ClassManager from ..orm.instrumentation import InstrumentationFactory INSTRUMENTATION_MANAGER = "__sa_instrumentation_manager__" """Attribute, elects custom instrumentation when present on a mapped class. Allows a class to specify a slightly or wildly different technique for tracking changes made to mapped attributes and collections. Only one instrumentation implementation is allowed in a given object inheritance hierarchy. The value of this attribute must be a callable and will be passed a class object. The callable must return one of: - An instance of an :class:`.InstrumentationManager` or subclass - An object implementing all or some of InstrumentationManager (TODO) - A dictionary of callables, implementing all or some of the above (TODO) - An instance of a :class:`.ClassManager` or subclass This attribute is consulted by SQLAlchemy instrumentation resolution, once the :mod:`sqlalchemy.ext.instrumentation` module has been imported. If custom finders are installed in the global instrumentation_finders list, they may or may not choose to honor this attribute. """ def find_native_user_instrumentation_hook(cls): """Find user-specified instrumentation management for a class.""" return getattr(cls, INSTRUMENTATION_MANAGER, None) instrumentation_finders = [find_native_user_instrumentation_hook] """An extensible sequence of callables which return instrumentation implementations When a class is registered, each callable will be passed a class object. If None is returned, the next finder in the sequence is consulted. Otherwise the return must be an instrumentation factory that follows the same guidelines as sqlalchemy.ext.instrumentation.INSTRUMENTATION_MANAGER. By default, the only finder is find_native_user_instrumentation_hook, which searches for INSTRUMENTATION_MANAGER. If all finders return None, standard ClassManager instrumentation is used. """ class ExtendedInstrumentationRegistry(InstrumentationFactory): """Extends :class:`.InstrumentationFactory` with additional bookkeeping, to accommodate multiple types of class managers. """ _manager_finders = weakref.WeakKeyDictionary() _state_finders = weakref.WeakKeyDictionary() _dict_finders = weakref.WeakKeyDictionary() _extended = False def _locate_extended_factory(self, class_): for finder in instrumentation_finders: factory = finder(class_) if factory is not None: manager = self._extended_class_manager(class_, factory) return manager, factory else: return None, None def _check_conflicts(self, class_, factory): existing_factories = self._collect_management_factories_for( class_ ).difference([factory]) if existing_factories: raise TypeError( "multiple instrumentation implementations specified " "in %s inheritance hierarchy: %r" % (class_.__name__, list(existing_factories)) ) def _extended_class_manager(self, class_, factory): manager = factory(class_) if not isinstance(manager, ClassManager): manager = _ClassInstrumentationAdapter(class_, manager) if factory != ClassManager and not self._extended: # somebody invoked a custom ClassManager. # reinstall global "getter" functions with the more # expensive ones. self._extended = True _install_instrumented_lookups() self._manager_finders[class_] = manager.manager_getter() self._state_finders[class_] = manager.state_getter() self._dict_finders[class_] = manager.dict_getter() return manager def _collect_management_factories_for(self, cls): """Return a collection of factories in play or specified for a hierarchy. Traverses the entire inheritance graph of a cls and returns a collection of instrumentation factories for those classes. Factories are extracted from active ClassManagers, if available, otherwise instrumentation_finders is consulted. """ hierarchy = util.class_hierarchy(cls) factories = set() for member in hierarchy: manager = self.manager_of_class(member) if manager is not None: factories.add(manager.factory) else: for finder in instrumentation_finders: factory = finder(member) if factory is not None: break else: factory = None factories.add(factory) factories.discard(None) return factories def unregister(self, class_): if class_ in self._manager_finders: del self._manager_finders[class_] del self._state_finders[class_] del self._dict_finders[class_] super(ExtendedInstrumentationRegistry, self).unregister(class_) def manager_of_class(self, cls): if cls is None: return None try: finder = self._manager_finders.get(cls, _default_manager_getter) except TypeError: # due to weakref lookup on invalid object return None else: return finder(cls) def state_of(self, instance): if instance is None: raise AttributeError("None has no persistent state.") return self._state_finders.get( instance.__class__, _default_state_getter )(instance) def dict_of(self, instance): if instance is None: raise AttributeError("None has no persistent state.") return self._dict_finders.get( instance.__class__, _default_dict_getter )(instance) orm_instrumentation._instrumentation_factory = ( _instrumentation_factory ) = ExtendedInstrumentationRegistry() orm_instrumentation.instrumentation_finders = instrumentation_finders class InstrumentationManager(object): """User-defined class instrumentation extension. :class:`.InstrumentationManager` can be subclassed in order to change how class instrumentation proceeds. This class exists for the purposes of integration with other object management frameworks which would like to entirely modify the instrumentation methodology of the ORM, and is not intended for regular usage. For interception of class instrumentation events, see :class:`.InstrumentationEvents`. The API for this class should be considered as semi-stable, and may change slightly with new releases. """ # r4361 added a mandatory (cls) constructor to this interface. # given that, perhaps class_ should be dropped from all of these # signatures. def __init__(self, class_): pass def manage(self, class_, manager): setattr(class_, "_default_class_manager", manager) def dispose(self, class_, manager): delattr(class_, "_default_class_manager") def manager_getter(self, class_): def get(cls): return cls._default_class_manager return get def instrument_attribute(self, class_, key, inst): pass def post_configure_attribute(self, class_, key, inst): pass def install_descriptor(self, class_, key, inst): setattr(class_, key, inst) def uninstall_descriptor(self, class_, key): delattr(class_, key) def install_member(self, class_, key, implementation): setattr(class_, key, implementation) def uninstall_member(self, class_, key): delattr(class_, key) def instrument_collection_class(self, class_, key, collection_class): return collections.prepare_instrumentation(collection_class) def get_instance_dict(self, class_, instance): return instance.__dict__ def initialize_instance_dict(self, class_, instance): pass def install_state(self, class_, instance, state): setattr(instance, "_default_state", state) def remove_state(self, class_, instance): delattr(instance, "_default_state") def state_getter(self, class_): return lambda instance: getattr(instance, "_default_state") def dict_getter(self, class_): return lambda inst: self.get_instance_dict(class_, inst) class _ClassInstrumentationAdapter(ClassManager): """Adapts a user-defined InstrumentationManager to a ClassManager.""" def __init__(self, class_, override): self._adapted = override self._get_state = self._adapted.state_getter(class_) self._get_dict = self._adapted.dict_getter(class_) ClassManager.__init__(self, class_) def manage(self): self._adapted.manage(self.class_, self) def dispose(self): self._adapted.dispose(self.class_) def manager_getter(self): return self._adapted.manager_getter(self.class_) def instrument_attribute(self, key, inst, propagated=False): ClassManager.instrument_attribute(self, key, inst, propagated) if not propagated: self._adapted.instrument_attribute(self.class_, key, inst) def post_configure_attribute(self, key): super(_ClassInstrumentationAdapter, self).post_configure_attribute(key) self._adapted.post_configure_attribute(self.class_, key, self[key]) def install_descriptor(self, key, inst): self._adapted.install_descriptor(self.class_, key, inst) def uninstall_descriptor(self, key): self._adapted.uninstall_descriptor(self.class_, key) def install_member(self, key, implementation): self._adapted.install_member(self.class_, key, implementation) def uninstall_member(self, key): self._adapted.uninstall_member(self.class_, key) def instrument_collection_class(self, key, collection_class): return self._adapted.instrument_collection_class( self.class_, key, collection_class ) def initialize_collection(self, key, state, factory): delegate = getattr(self._adapted, "initialize_collection", None) if delegate: return delegate(key, state, factory) else: return ClassManager.initialize_collection( self, key, state, factory ) def new_instance(self, state=None): instance = self.class_.__new__(self.class_) self.setup_instance(instance, state) return instance def _new_state_if_none(self, instance): """Install a default InstanceState if none is present. A private convenience method used by the __init__ decorator. """ if self.has_state(instance): return False else: return self.setup_instance(instance) def setup_instance(self, instance, state=None): self._adapted.initialize_instance_dict(self.class_, instance) if state is None: state = self._state_constructor(instance, self) # the given instance is assumed to have no state self._adapted.install_state(self.class_, instance, state) return state def teardown_instance(self, instance): self._adapted.remove_state(self.class_, instance) def has_state(self, instance): try: self._get_state(instance) except orm_exc.NO_STATE: return False else: return True def state_getter(self): return self._get_state def dict_getter(self): return self._get_dict def _install_instrumented_lookups(): """Replace global class/object management functions with ExtendedInstrumentationRegistry implementations, which allow multiple types of class managers to be present, at the cost of performance. This function is called only by ExtendedInstrumentationRegistry and unit tests specific to this behavior. The _reinstall_default_lookups() function can be called after this one to re-establish the default functions. """ _install_lookups( dict( instance_state=_instrumentation_factory.state_of, instance_dict=_instrumentation_factory.dict_of, manager_of_class=_instrumentation_factory.manager_of_class, ) ) def _reinstall_default_lookups(): """Restore simplified lookups.""" _install_lookups( dict( instance_state=_default_state_getter, instance_dict=_default_dict_getter, manager_of_class=_default_manager_getter, ) ) _instrumentation_factory._extended = False def _install_lookups(lookups): global instance_state, instance_dict, manager_of_class instance_state = lookups["instance_state"] instance_dict = lookups["instance_dict"] manager_of_class = lookups["manager_of_class"] orm_base.instance_state = ( attributes.instance_state ) = orm_instrumentation.instance_state = instance_state orm_base.instance_dict = ( attributes.instance_dict ) = orm_instrumentation.instance_dict = instance_dict orm_base.manager_of_class = ( attributes.manager_of_class ) = orm_instrumentation.manager_of_class = manager_of_class	14.03 KB	2026-01-20 12:58
📄 mutable.py # ext/mutable.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php r"""Provide support for tracking of in-place changes to scalar values, which are propagated into ORM change events on owning parent objects. .. _mutable_scalars: Establishing Mutability on Scalar Column Values =============================================== A typical example of a "mutable" structure is a Python dictionary. Following the example introduced in :ref:`types_toplevel`, we begin with a custom type that marshals Python dictionaries into JSON strings before being persisted:: from sqlalchemy.types import TypeDecorator, VARCHAR import json class JSONEncodedDict(TypeDecorator): "Represents an immutable structure as a json-encoded string." impl = VARCHAR def process_bind_param(self, value, dialect): if value is not None: value = json.dumps(value) return value def process_result_value(self, value, dialect): if value is not None: value = json.loads(value) return value The usage of ``json`` is only for the purposes of example. The :mod:`sqlalchemy.ext.mutable` extension can be used with any type whose target Python type may be mutable, including :class:`.PickleType`, :class:`_postgresql.ARRAY`, etc. When using the :mod:`sqlalchemy.ext.mutable` extension, the value itself tracks all parents which reference it. Below, we illustrate a simple version of the :class:`.MutableDict` dictionary object, which applies the :class:`.Mutable` mixin to a plain Python dictionary:: from sqlalchemy.ext.mutable import Mutable class MutableDict(Mutable, dict): @classmethod def coerce(cls, key, value): "Convert plain dictionaries to MutableDict." if not isinstance(value, MutableDict): if isinstance(value, dict): return MutableDict(value) # this call will raise ValueError return Mutable.coerce(key, value) else: return value def __setitem__(self, key, value): "Detect dictionary set events and emit change events." dict.__setitem__(self, key, value) self.changed() def __delitem__(self, key): "Detect dictionary del events and emit change events." dict.__delitem__(self, key) self.changed() The above dictionary class takes the approach of subclassing the Python built-in ``dict`` to produce a dict subclass which routes all mutation events through ``__setitem__``. There are variants on this approach, such as subclassing ``UserDict.UserDict`` or ``collections.MutableMapping``; the part that's important to this example is that the :meth:`.Mutable.changed` method is called whenever an in-place change to the datastructure takes place. We also redefine the :meth:`.Mutable.coerce` method which will be used to convert any values that are not instances of ``MutableDict``, such as the plain dictionaries returned by the ``json`` module, into the appropriate type. Defining this method is optional; we could just as well created our ``JSONEncodedDict`` such that it always returns an instance of ``MutableDict``, and additionally ensured that all calling code uses ``MutableDict`` explicitly. When :meth:`.Mutable.coerce` is not overridden, any values applied to a parent object which are not instances of the mutable type will raise a ``ValueError``. Our new ``MutableDict`` type offers a class method :meth:`~.Mutable.as_mutable` which we can use within column metadata to associate with types. This method grabs the given type object or class and associates a listener that will detect all future mappings of this type, applying event listening instrumentation to the mapped attribute. Such as, with classical table metadata:: from sqlalchemy import Table, Column, Integer my_data = Table('my_data', metadata, Column('id', Integer, primary_key=True), Column('data', MutableDict.as_mutable(JSONEncodedDict)) ) Above, :meth:`~.Mutable.as_mutable` returns an instance of ``JSONEncodedDict`` (if the type object was not an instance already), which will intercept any attributes which are mapped against this type. Below we establish a simple mapping against the ``my_data`` table:: from sqlalchemy import mapper class MyDataClass(object): pass # associates mutation listeners with MyDataClass.data mapper(MyDataClass, my_data) The ``MyDataClass.data`` member will now be notified of in place changes to its value. There's no difference in usage when using declarative:: from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class MyDataClass(Base): __tablename__ = 'my_data' id = Column(Integer, primary_key=True) data = Column(MutableDict.as_mutable(JSONEncodedDict)) Any in-place changes to the ``MyDataClass.data`` member will flag the attribute as "dirty" on the parent object:: >>> from sqlalchemy.orm import Session >>> sess = Session() >>> m1 = MyDataClass(data={'value1':'foo'}) >>> sess.add(m1) >>> sess.commit() >>> m1.data['value1'] = 'bar' >>> assert m1 in sess.dirty True The ``MutableDict`` can be associated with all future instances of ``JSONEncodedDict`` in one step, using :meth:`~.Mutable.associate_with`. This is similar to :meth:`~.Mutable.as_mutable` except it will intercept all occurrences of ``MutableDict`` in all mappings unconditionally, without the need to declare it individually:: MutableDict.associate_with(JSONEncodedDict) class MyDataClass(Base): __tablename__ = 'my_data' id = Column(Integer, primary_key=True) data = Column(JSONEncodedDict) Supporting Pickling -------------------- The key to the :mod:`sqlalchemy.ext.mutable` extension relies upon the placement of a ``weakref.WeakKeyDictionary`` upon the value object, which stores a mapping of parent mapped objects keyed to the attribute name under which they are associated with this value. ``WeakKeyDictionary`` objects are not picklable, due to the fact that they contain weakrefs and function callbacks. In our case, this is a good thing, since if this dictionary were picklable, it could lead to an excessively large pickle size for our value objects that are pickled by themselves outside of the context of the parent. The developer responsibility here is only to provide a ``__getstate__`` method that excludes the :meth:`~MutableBase._parents` collection from the pickle stream:: class MyMutableType(Mutable): def __getstate__(self): d = self.__dict__.copy() d.pop('_parents', None) return d With our dictionary example, we need to return the contents of the dict itself (and also restore them on __setstate__):: class MutableDict(Mutable, dict): # .... def __getstate__(self): return dict(self) def __setstate__(self, state): self.update(state) In the case that our mutable value object is pickled as it is attached to one or more parent objects that are also part of the pickle, the :class:`.Mutable` mixin will re-establish the :attr:`.Mutable._parents` collection on each value object as the owning parents themselves are unpickled. Receiving Events ---------------- The :meth:`.AttributeEvents.modified` event handler may be used to receive an event when a mutable scalar emits a change event. This event handler is called when the :func:`.attributes.flag_modified` function is called from within the mutable extension:: from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import event Base = declarative_base() class MyDataClass(Base): __tablename__ = 'my_data' id = Column(Integer, primary_key=True) data = Column(MutableDict.as_mutable(JSONEncodedDict)) @event.listens_for(MyDataClass.data, "modified") def modified_json(instance): print("json value modified:", instance.data) .. _mutable_composites: Establishing Mutability on Composites ===================================== Composites are a special ORM feature which allow a single scalar attribute to be assigned an object value which represents information "composed" from one or more columns from the underlying mapped table. The usual example is that of a geometric "point", and is introduced in :ref:`mapper_composite`. As is the case with :class:`.Mutable`, the user-defined composite class subclasses :class:`.MutableComposite` as a mixin, and detects and delivers change events to its parents via the :meth:`.MutableComposite.changed` method. In the case of a composite class, the detection is usually via the usage of Python descriptors (i.e. ``@property``), or alternatively via the special Python method ``__setattr__()``. Below we expand upon the ``Point`` class introduced in :ref:`mapper_composite` to subclass :class:`.MutableComposite` and to also route attribute set events via ``__setattr__`` to the :meth:`.MutableComposite.changed` method:: from sqlalchemy.ext.mutable import MutableComposite class Point(MutableComposite): def __init__(self, x, y): self.x = x self.y = y def __setattr__(self, key, value): "Intercept set events" # set the attribute object.__setattr__(self, key, value) # alert all parents to the change self.changed() def __composite_values__(self): return self.x, self.y def __eq__(self, other): return isinstance(other, Point) and \ other.x == self.x and \ other.y == self.y def __ne__(self, other): return not self.__eq__(other) The :class:`.MutableComposite` class uses a Python metaclass to automatically establish listeners for any usage of :func:`_orm.composite` that specifies our ``Point`` type. Below, when ``Point`` is mapped to the ``Vertex`` class, listeners are established which will route change events from ``Point`` objects to each of the ``Vertex.start`` and ``Vertex.end`` attributes:: from sqlalchemy.orm import composite, mapper from sqlalchemy import Table, Column vertices = Table('vertices', metadata, Column('id', Integer, primary_key=True), Column('x1', Integer), Column('y1', Integer), Column('x2', Integer), Column('y2', Integer), ) class Vertex(object): pass mapper(Vertex, vertices, properties={ 'start': composite(Point, vertices.c.x1, vertices.c.y1), 'end': composite(Point, vertices.c.x2, vertices.c.y2) }) Any in-place changes to the ``Vertex.start`` or ``Vertex.end`` members will flag the attribute as "dirty" on the parent object:: >>> from sqlalchemy.orm import Session >>> sess = Session() >>> v1 = Vertex(start=Point(3, 4), end=Point(12, 15)) >>> sess.add(v1) >>> sess.commit() >>> v1.end.x = 8 >>> assert v1 in sess.dirty True Coercing Mutable Composites --------------------------- The :meth:`.MutableBase.coerce` method is also supported on composite types. In the case of :class:`.MutableComposite`, the :meth:`.MutableBase.coerce` method is only called for attribute set operations, not load operations. Overriding the :meth:`.MutableBase.coerce` method is essentially equivalent to using a :func:`.validates` validation routine for all attributes which make use of the custom composite type:: class Point(MutableComposite): # other Point methods # ... def coerce(cls, key, value): if isinstance(value, tuple): value = Point(value) elif not isinstance(value, Point): raise ValueError("tuple or Point expected") return value Supporting Pickling -------------------- As is the case with :class:`.Mutable`, the :class:`.MutableComposite` helper class uses a ``weakref.WeakKeyDictionary`` available via the :meth:`MutableBase._parents` attribute which isn't picklable. If we need to pickle instances of ``Point`` or its owning class ``Vertex``, we at least need to define a ``__getstate__`` that doesn't include the ``_parents`` dictionary. Below we define both a ``__getstate__`` and a ``__setstate__`` that package up the minimal form of our ``Point`` class:: class Point(MutableComposite): # ... def __getstate__(self): return self.x, self.y def __setstate__(self, state): self.x, self.y = state As with :class:`.Mutable`, the :class:`.MutableComposite` augments the pickling process of the parent's object-relational state so that the :meth:`MutableBase._parents` collection is restored to all ``Point`` objects. """ import weakref from .. import event from .. import types from ..orm import Mapper from ..orm import mapper from ..orm import object_mapper from ..orm.attributes import flag_modified from ..sql.base import SchemaEventTarget from ..util import memoized_property class MutableBase(object): """Common base class to :class:`.Mutable` and :class:`.MutableComposite`. """ @memoized_property def _parents(self): """Dictionary of parent object->attribute name on the parent. This attribute is a so-called "memoized" property. It initializes itself with a new ``weakref.WeakKeyDictionary`` the first time it is accessed, returning the same object upon subsequent access. """ return weakref.WeakKeyDictionary() @classmethod def coerce(cls, key, value): """Given a value, coerce it into the target type. Can be overridden by custom subclasses to coerce incoming data into a particular type. By default, raises ``ValueError``. This method is called in different scenarios depending on if the parent class is of type :class:`.Mutable` or of type :class:`.MutableComposite`. In the case of the former, it is called for both attribute-set operations as well as during ORM loading operations. For the latter, it is only called during attribute-set operations; the mechanics of the :func:`.composite` construct handle coercion during load operations. :param key: string name of the ORM-mapped attribute being set. :param value: the incoming value. :return: the method should return the coerced value, or raise ``ValueError`` if the coercion cannot be completed. """ if value is None: return None msg = "Attribute '%s' does not accept objects of type %s" raise ValueError(msg % (key, type(value))) @classmethod def _get_listen_keys(cls, attribute): """Given a descriptor attribute, return a ``set()`` of the attribute keys which indicate a change in the state of this attribute. This is normally just ``set([attribute.key])``, but can be overridden to provide for additional keys. E.g. a :class:`.MutableComposite` augments this set with the attribute keys associated with the columns that comprise the composite value. This collection is consulted in the case of intercepting the :meth:`.InstanceEvents.refresh` and :meth:`.InstanceEvents.refresh_flush` events, which pass along a list of attribute names that have been refreshed; the list is compared against this set to determine if action needs to be taken. .. versionadded:: 1.0.5 """ return {attribute.key} @classmethod def _listen_on_attribute(cls, attribute, coerce, parent_cls): """Establish this type as a mutation listener for the given mapped descriptor. """ key = attribute.key if parent_cls is not attribute.class_: return # rely on "propagate" here parent_cls = attribute.class_ listen_keys = cls._get_listen_keys(attribute) def load(state, args): """Listen for objects loaded or refreshed. Wrap the target data member's value with ``Mutable``. """ val = state.dict.get(key, None) if val is not None: if coerce: val = cls.coerce(key, val) state.dict[key] = val val._parents[state.obj()] = key def load_attrs(state, ctx, attrs): if not attrs or listen_keys.intersection(attrs): load(state) def set_(target, value, oldvalue, initiator): """Listen for set/replace events on the target data member. Establish a weak reference to the parent object on the incoming value, remove it for the one outgoing. """ if value is oldvalue: return value if not isinstance(value, cls): value = cls.coerce(key, value) if value is not None: value._parents[target.obj()] = key if isinstance(oldvalue, cls): oldvalue._parents.pop(target.obj(), None) return value def pickle(state, state_dict): val = state.dict.get(key, None) if val is not None: if "ext.mutable.values" not in state_dict: state_dict["ext.mutable.values"] = [] state_dict["ext.mutable.values"].append(val) def unpickle(state, state_dict): if "ext.mutable.values" in state_dict: for val in state_dict["ext.mutable.values"]: val._parents[state.obj()] = key event.listen(parent_cls, "load", load, raw=True, propagate=True) event.listen( parent_cls, "refresh", load_attrs, raw=True, propagate=True ) event.listen( parent_cls, "refresh_flush", load_attrs, raw=True, propagate=True ) event.listen( attribute, "set", set_, raw=True, retval=True, propagate=True ) event.listen(parent_cls, "pickle", pickle, raw=True, propagate=True) event.listen( parent_cls, "unpickle", unpickle, raw=True, propagate=True ) class Mutable(MutableBase): """Mixin that defines transparent propagation of change events to a parent object. See the example in :ref:`mutable_scalars` for usage information. """ def changed(self): """Subclasses should call this method whenever change events occur.""" for parent, key in self._parents.items(): flag_modified(parent, key) @classmethod def associate_with_attribute(cls, attribute): """Establish this type as a mutation listener for the given mapped descriptor. """ cls._listen_on_attribute(attribute, True, attribute.class_) @classmethod def associate_with(cls, sqltype): """Associate this wrapper with all future mapped columns of the given type. This is a convenience method that calls ``associate_with_attribute`` automatically. .. warning:: The listeners established by this method are global to all mappers, and are not garbage collected. Only use :meth:`.associate_with` for types that are permanent to an application, not with ad-hoc types else this will cause unbounded growth in memory usage. """ def listen_for_type(mapper, class_): if mapper.non_primary: return for prop in mapper.column_attrs: if isinstance(prop.columns[0].type, sqltype): cls.associate_with_attribute(getattr(class_, prop.key)) event.listen(mapper, "mapper_configured", listen_for_type) @classmethod def as_mutable(cls, sqltype): """Associate a SQL type with this mutable Python type. This establishes listeners that will detect ORM mappings against the given type, adding mutation event trackers to those mappings. The type is returned, unconditionally as an instance, so that :meth:`.as_mutable` can be used inline:: Table('mytable', metadata, Column('id', Integer, primary_key=True), Column('data', MyMutableType.as_mutable(PickleType)) ) Note that the returned type is always an instance, even if a class is given, and that only columns which are declared specifically with that type instance receive additional instrumentation. To associate a particular mutable type with all occurrences of a particular type, use the :meth:`.Mutable.associate_with` classmethod of the particular :class:`.Mutable` subclass to establish a global association. .. warning:: The listeners established by this method are global to all mappers, and are not garbage collected. Only use :meth:`.as_mutable` for types that are permanent to an application, not with ad-hoc types else this will cause unbounded growth in memory usage. """ sqltype = types.to_instance(sqltype) # a SchemaType will be copied when the Column is copied, # and we'll lose our ability to link that type back to the original. # so track our original type w/ columns if isinstance(sqltype, SchemaEventTarget): @event.listens_for(sqltype, "before_parent_attach") def _add_column_memo(sqltyp, parent): parent.info["_ext_mutable_orig_type"] = sqltyp schema_event_check = True else: schema_event_check = False def listen_for_type(mapper, class_): if mapper.non_primary: return for prop in mapper.column_attrs: if ( schema_event_check and hasattr(prop.expression, "info") and prop.expression.info.get("_ext_mutable_orig_type") is sqltype ) or (prop.columns[0].type is sqltype): cls.associate_with_attribute(getattr(class_, prop.key)) event.listen(mapper, "mapper_configured", listen_for_type) return sqltype class MutableComposite(MutableBase): """Mixin that defines transparent propagation of change events on a SQLAlchemy "composite" object to its owning parent or parents. See the example in :ref:`mutable_composites` for usage information. """ @classmethod def _get_listen_keys(cls, attribute): return {attribute.key}.union(attribute.property._attribute_keys) def changed(self): """Subclasses should call this method whenever change events occur.""" for parent, key in self._parents.items(): prop = object_mapper(parent).get_property(key) for value, attr_name in zip( self.__composite_values__(), prop._attribute_keys ): setattr(parent, attr_name, value) def _setup_composite_listener(): def _listen_for_type(mapper, class_): for prop in mapper.iterate_properties: if ( hasattr(prop, "composite_class") and isinstance(prop.composite_class, type) and issubclass(prop.composite_class, MutableComposite) ): prop.composite_class._listen_on_attribute( getattr(class_, prop.key), False, class_ ) if not event.contains(Mapper, "mapper_configured", _listen_for_type): event.listen(Mapper, "mapper_configured", _listen_for_type) _setup_composite_listener() class MutableDict(Mutable, dict): """A dictionary type that implements :class:`.Mutable`. The :class:`.MutableDict` object implements a dictionary that will emit change events to the underlying mapping when the contents of the dictionary are altered, including when values are added or removed. Note that :class:`.MutableDict` does not apply mutable tracking to the values themselves inside the dictionary. Therefore it is not a sufficient solution for the use case of tracking deep changes to a recursive dictionary structure, such as a JSON structure. To support this use case, build a subclass of :class:`.MutableDict` that provides appropriate coercion to the values placed in the dictionary so that they too are "mutable", and emit events up to their parent structure. .. seealso:: :class:`.MutableList` :class:`.MutableSet` """ def __setitem__(self, key, value): """Detect dictionary set events and emit change events.""" dict.__setitem__(self, key, value) self.changed() def setdefault(self, key, value): result = dict.setdefault(self, key, value) self.changed() return result def __delitem__(self, key): """Detect dictionary del events and emit change events.""" dict.__delitem__(self, key) self.changed() def update(self, a, kw): dict.update(self, a, *kw) self.changed() def pop(self, arg): result = dict.pop(self, arg) self.changed() return result def popitem(self): result = dict.popitem(self) self.changed() return result def clear(self): dict.clear(self) self.changed() @classmethod def coerce(cls, key, value): """Convert plain dictionary to instance of this class.""" if not isinstance(value, cls): if isinstance(value, dict): return cls(value) return Mutable.coerce(key, value) else: return value def __getstate__(self): return dict(self) def __setstate__(self, state): self.update(state) class MutableList(Mutable, list): """A list type that implements :class:`.Mutable`. The :class:`.MutableList` object implements a list that will emit change events to the underlying mapping when the contents of the list are altered, including when values are added or removed. Note that :class:`.MutableList` does not* apply mutable tracking to the values themselves inside the list. Therefore it is not a sufficient solution for the use case of tracking deep changes to a recursive mutable structure, such as a JSON structure. To support this use case, build a subclass of :class:`.MutableList` that provides appropriate coercion to the values placed in the dictionary so that they too are "mutable", and emit events up to their parent structure. .. versionadded:: 1.1 .. seealso:: :class:`.MutableDict` :class:`.MutableSet` """ def __reduce_ex__(self, proto): return (self.__class__, (list(self),)) # needed for backwards compatibility with # older pickles def __setstate__(self, state): self[:] = state def __setitem__(self, index, value): """Detect list set events and emit change events.""" list.__setitem__(self, index, value) self.changed() def __setslice__(self, start, end, value): """Detect list set events and emit change events.""" list.__setslice__(self, start, end, value) self.changed() def __delitem__(self, index): """Detect list del events and emit change events.""" list.__delitem__(self, index) self.changed() def __delslice__(self, start, end): """Detect list del events and emit change events.""" list.__delslice__(self, start, end) self.changed() def pop(self, arg): result = list.pop(self, arg) self.changed() return result def append(self, x): list.append(self, x) self.changed() def extend(self, x): list.extend(self, x) self.changed() def __iadd__(self, x): self.extend(x) return self def insert(self, i, x): list.insert(self, i, x) self.changed() def remove(self, i): list.remove(self, i) self.changed() def clear(self): list.clear(self) self.changed() def sort(self, kw): list.sort(self, kw) self.changed() def reverse(self): list.reverse(self) self.changed() @classmethod def coerce(cls, index, value): """Convert plain list to instance of this class.""" if not isinstance(value, cls): if isinstance(value, list): return cls(value) return Mutable.coerce(index, value) else: return value class MutableSet(Mutable, set): """A set type that implements :class:`.Mutable`. The :class:`.MutableSet` object implements a set that will emit change events to the underlying mapping when the contents of the set are altered, including when values are added or removed. Note that :class:`.MutableSet` does not apply mutable tracking to the values themselves inside the set. Therefore it is not a sufficient solution for the use case of tracking deep changes to a recursive mutable structure. To support this use case, build a subclass of :class:`.MutableSet` that provides appropriate coercion to the values placed in the dictionary so that they too are "mutable", and emit events up to their parent structure. .. versionadded:: 1.1 .. seealso:: :class:`.MutableDict` :class:`.MutableList` """ def update(self, arg): set.update(self, arg) self.changed() def intersection_update(self, arg): set.intersection_update(self, arg) self.changed() def difference_update(self, arg): set.difference_update(self, arg) self.changed() def symmetric_difference_update(self, arg): set.symmetric_difference_update(self, arg) self.changed() def __ior__(self, other): self.update(other) return self def __iand__(self, other): self.intersection_update(other) return self def __ixor__(self, other): self.symmetric_difference_update(other) return self def __isub__(self, other): self.difference_update(other) return self def add(self, elem): set.add(self, elem) self.changed() def remove(self, elem): set.remove(self, elem) self.changed() def discard(self, elem): set.discard(self, elem) self.changed() def pop(self, arg): result = set.pop(self, arg) self.changed() return result def clear(self): set.clear(self) self.changed() @classmethod def coerce(cls, index, value): """Convert plain set to instance of this class.""" if not isinstance(value, cls): if isinstance(value, set): return cls(value) return Mutable.coerce(index, value) else: return value def __getstate__(self): return set(self) def __setstate__(self, state): self.update(state) def __reduce_ex__(self, proto): return (self.__class__, (list(self),))	31.07 KB	2026-01-20 12:58
📄 orderinglist.py # ext/orderinglist.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """A custom list that manages index/position information for contained elements. :author: Jason Kirtland ``orderinglist`` is a helper for mutable ordered relationships. It will intercept list operations performed on a :func:`_orm.relationship`-managed collection and automatically synchronize changes in list position onto a target scalar attribute. Example: A ``slide`` table, where each row refers to zero or more entries in a related ``bullet`` table. The bullets within a slide are displayed in order based on the value of the ``position`` column in the ``bullet`` table. As entries are reordered in memory, the value of the ``position`` attribute should be updated to reflect the new sort order:: Base = declarative_base() class Slide(Base): __tablename__ = 'slide' id = Column(Integer, primary_key=True) name = Column(String) bullets = relationship("Bullet", order_by="Bullet.position") class Bullet(Base): __tablename__ = 'bullet' id = Column(Integer, primary_key=True) slide_id = Column(Integer, ForeignKey('slide.id')) position = Column(Integer) text = Column(String) The standard relationship mapping will produce a list-like attribute on each ``Slide`` containing all related ``Bullet`` objects, but coping with changes in ordering is not handled automatically. When appending a ``Bullet`` into ``Slide.bullets``, the ``Bullet.position`` attribute will remain unset until manually assigned. When the ``Bullet`` is inserted into the middle of the list, the following ``Bullet`` objects will also need to be renumbered. The :class:`.OrderingList` object automates this task, managing the ``position`` attribute on all ``Bullet`` objects in the collection. It is constructed using the :func:`.ordering_list` factory:: from sqlalchemy.ext.orderinglist import ordering_list Base = declarative_base() class Slide(Base): __tablename__ = 'slide' id = Column(Integer, primary_key=True) name = Column(String) bullets = relationship("Bullet", order_by="Bullet.position", collection_class=ordering_list('position')) class Bullet(Base): __tablename__ = 'bullet' id = Column(Integer, primary_key=True) slide_id = Column(Integer, ForeignKey('slide.id')) position = Column(Integer) text = Column(String) With the above mapping the ``Bullet.position`` attribute is managed:: s = Slide() s.bullets.append(Bullet()) s.bullets.append(Bullet()) s.bullets[1].position >>> 1 s.bullets.insert(1, Bullet()) s.bullets[2].position >>> 2 The :class:`.OrderingList` construct only works with changes to a collection, and not the initial load from the database, and requires that the list be sorted when loaded. Therefore, be sure to specify ``order_by`` on the :func:`_orm.relationship` against the target ordering attribute, so that the ordering is correct when first loaded. .. warning:: :class:`.OrderingList` only provides limited functionality when a primary key column or unique column is the target of the sort. Operations that are unsupported or are problematic include: * two entries must trade values. This is not supported directly in the case of a primary key or unique constraint because it means at least one row would need to be temporarily removed first, or changed to a third, neutral value while the switch occurs. * an entry must be deleted in order to make room for a new entry. SQLAlchemy's unit of work performs all INSERTs before DELETEs within a single flush. In the case of a primary key, it will trade an INSERT/DELETE of the same primary key for an UPDATE statement in order to lessen the impact of this limitation, however this does not take place for a UNIQUE column. A future feature will allow the "DELETE before INSERT" behavior to be possible, alleviating this limitation, though this feature will require explicit configuration at the mapper level for sets of columns that are to be handled in this way. :func:`.ordering_list` takes the name of the related object's ordering attribute as an argument. By default, the zero-based integer index of the object's position in the :func:`.ordering_list` is synchronized with the ordering attribute: index 0 will get position 0, index 1 position 1, etc. To start numbering at 1 or some other integer, provide ``count_from=1``. """ from .. import util from ..orm.collections import collection from ..orm.collections import collection_adapter __all__ = ["ordering_list"] def ordering_list(attr, count_from=None, kw): """Prepares an :class:`OrderingList` factory for use in mapper definitions. Returns an object suitable for use as an argument to a Mapper relationship's ``collection_class`` option. e.g.:: from sqlalchemy.ext.orderinglist import ordering_list class Slide(Base): __tablename__ = 'slide' id = Column(Integer, primary_key=True) name = Column(String) bullets = relationship("Bullet", order_by="Bullet.position", collection_class=ordering_list('position')) :param attr: Name of the mapped attribute to use for storage and retrieval of ordering information :param count_from: Set up an integer-based ordering, starting at ``count_from``. For example, ``ordering_list('pos', count_from=1)`` would create a 1-based list in SQL, storing the value in the 'pos' column. Ignored if ``ordering_func`` is supplied. Additional arguments are passed to the :class:`.OrderingList` constructor. """ kw = _unsugar_count_from(count_from=count_from, kw) return lambda: OrderingList(attr, kw) # Ordering utility functions def count_from_0(index, collection): """Numbering function: consecutive integers starting at 0.""" return index def count_from_1(index, collection): """Numbering function: consecutive integers starting at 1.""" return index + 1 def count_from_n_factory(start): """Numbering function: consecutive integers starting at arbitrary start.""" def f(index, collection): return index + start try: f.__name__ = "count_from_%i" % start except TypeError: pass return f def _unsugar_count_from(kw): """Builds counting functions from keyword arguments. Keyword argument filter, prepares a simple ``ordering_func`` from a ``count_from`` argument, otherwise passes ``ordering_func`` on unchanged. """ count_from = kw.pop("count_from", None) if kw.get("ordering_func", None) is None and count_from is not None: if count_from == 0: kw["ordering_func"] = count_from_0 elif count_from == 1: kw["ordering_func"] = count_from_1 else: kw["ordering_func"] = count_from_n_factory(count_from) return kw class OrderingList(list): """A custom list that manages position information for its children. The :class:`.OrderingList` object is normally set up using the :func:`.ordering_list` factory function, used in conjunction with the :func:`_orm.relationship` function. """ def __init__( self, ordering_attr=None, ordering_func=None, reorder_on_append=False ): """A custom list that manages position information for its children. ``OrderingList`` is a ``collection_class`` list implementation that syncs position in a Python list with a position attribute on the mapped objects. This implementation relies on the list starting in the proper order, so be sure to put an ``order_by`` on your relationship. :param ordering_attr: Name of the attribute that stores the object's order in the relationship. :param ordering_func: Optional. A function that maps the position in the Python list to a value to store in the ``ordering_attr``. Values returned are usually (but need not be!) integers. An ``ordering_func`` is called with two positional parameters: the index of the element in the list, and the list itself. If omitted, Python list indexes are used for the attribute values. Two basic pre-built numbering functions are provided in this module: ``count_from_0`` and ``count_from_1``. For more exotic examples like stepped numbering, alphabetical and Fibonacci numbering, see the unit tests. :param reorder_on_append: Default False. When appending an object with an existing (non-None) ordering value, that value will be left untouched unless ``reorder_on_append`` is true. This is an optimization to avoid a variety of dangerous unexpected database writes. SQLAlchemy will add instances to the list via append() when your object loads. If for some reason the result set from the database skips a step in the ordering (say, row '1' is missing but you get '2', '3', and '4'), reorder_on_append=True would immediately renumber the items to '1', '2', '3'. If you have multiple sessions making changes, any of whom happen to load this collection even in passing, all of the sessions would try to "clean up" the numbering in their commits, possibly causing all but one to fail with a concurrent modification error. Recommend leaving this with the default of False, and just call ``reorder()`` if you're doing ``append()`` operations with previously ordered instances or when doing some housekeeping after manual sql operations. """ self.ordering_attr = ordering_attr if ordering_func is None: ordering_func = count_from_0 self.ordering_func = ordering_func self.reorder_on_append = reorder_on_append # More complex serialization schemes (multi column, e.g.) are possible by # subclassing and reimplementing these two methods. def _get_order_value(self, entity): return getattr(entity, self.ordering_attr) def _set_order_value(self, entity, value): setattr(entity, self.ordering_attr, value) def reorder(self): """Synchronize ordering for the entire collection. Sweeps through the list and ensures that each object has accurate ordering information set. """ for index, entity in enumerate(self): self._order_entity(index, entity, True) # As of 0.5, _reorder is no longer semi-private _reorder = reorder def _order_entity(self, index, entity, reorder=True): have = self._get_order_value(entity) # Don't disturb existing ordering if reorder is False if have is not None and not reorder: return should_be = self.ordering_func(index, self) if have != should_be: self._set_order_value(entity, should_be) def append(self, entity): super(OrderingList, self).append(entity) self._order_entity(len(self) - 1, entity, self.reorder_on_append) def _raw_append(self, entity): """Append without any ordering behavior.""" super(OrderingList, self).append(entity) _raw_append = collection.adds(1)(_raw_append) def insert(self, index, entity): super(OrderingList, self).insert(index, entity) self._reorder() def remove(self, entity): super(OrderingList, self).remove(entity) adapter = collection_adapter(self) if adapter and adapter._referenced_by_owner: self._reorder() def pop(self, index=-1): entity = super(OrderingList, self).pop(index) self._reorder() return entity def __setitem__(self, index, entity): if isinstance(index, slice): step = index.step or 1 start = index.start or 0 if start < 0: start += len(self) stop = index.stop or len(self) if stop < 0: stop += len(self) for i in range(start, stop, step): self.__setitem__(i, entity[i]) else: self._order_entity(index, entity, True) super(OrderingList, self).__setitem__(index, entity) def __delitem__(self, index): super(OrderingList, self).__delitem__(index) self._reorder() def __setslice__(self, start, end, values): super(OrderingList, self).__setslice__(start, end, values) self._reorder() def __delslice__(self, start, end): super(OrderingList, self).__delslice__(start, end) self._reorder() def __reduce__(self): return _reconstitute, (self.__class__, self.__dict__, list(self)) for func_name, func in list(locals().items()): if ( util.callable(func) and func.__name__ == func_name and not func.__doc__ and hasattr(list, func_name) ): func.__doc__ = getattr(list, func_name).__doc__ del func_name, func def _reconstitute(cls, dict_, items): """Reconstitute an :class:`.OrderingList`. This is the adjoint to :meth:`.OrderingList.__reduce__`. It is used for unpickling :class:`.OrderingList` objects. """ obj = cls.__new__(cls) obj.__dict__.update(dict_) list.extend(obj, items) return obj	13.57 KB	2026-01-20 12:58
📄 serializer.py # ext/serializer.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """Serializer/Deserializer objects for usage with SQLAlchemy query structures, allowing "contextual" deserialization. Any SQLAlchemy query structure, either based on sqlalchemy.sql.* or sqlalchemy.orm.* can be used. The mappers, Tables, Columns, Session etc. which are referenced by the structure are not persisted in serialized form, but are instead re-associated with the query structure when it is deserialized. Usage is nearly the same as that of the standard Python pickle module:: from sqlalchemy.ext.serializer import loads, dumps metadata = MetaData(bind=some_engine) Session = scoped_session(sessionmaker()) # ... define mappers query = Session.query(MyClass). filter(MyClass.somedata=='foo').order_by(MyClass.sortkey) # pickle the query serialized = dumps(query) # unpickle. Pass in metadata + scoped_session query2 = loads(serialized, metadata, Session) print query2.all() Similar restrictions as when using raw pickle apply; mapped classes must be themselves be pickleable, meaning they are importable from a module-level namespace. The serializer module is only appropriate for query structures. It is not needed for: * instances of user-defined classes. These contain no references to engines, sessions or expression constructs in the typical case and can be serialized directly. * Table metadata that is to be loaded entirely from the serialized structure (i.e. is not already declared in the application). Regular pickle.loads()/dumps() can be used to fully dump any ``MetaData`` object, typically one which was reflected from an existing database at some previous point in time. The serializer module is specifically for the opposite case, where the Table metadata is already present in memory. """ import re from .. import Column from .. import Table from ..engine import Engine from ..orm import class_mapper from ..orm.attributes import QueryableAttribute from ..orm.interfaces import MapperProperty from ..orm.mapper import Mapper from ..orm.session import Session from ..util import b64decode from ..util import b64encode from ..util import byte_buffer from ..util import pickle from ..util import text_type __all__ = ["Serializer", "Deserializer", "dumps", "loads"] def Serializer(args, kw): pickler = pickle.Pickler(args, *kw) def persistent_id(obj): # print "serializing:", repr(obj) if isinstance(obj, QueryableAttribute): cls = obj.impl.class_ key = obj.impl.key id_ = "attribute:" + key + ":" + b64encode(pickle.dumps(cls)) elif isinstance(obj, Mapper) and not obj.non_primary: id_ = "mapper:" + b64encode(pickle.dumps(obj.class_)) elif isinstance(obj, MapperProperty) and not obj.parent.non_primary: id_ = ( "mapperprop:" + b64encode(pickle.dumps(obj.parent.class_)) + ":" + obj.key ) elif isinstance(obj, Table): id_ = "table:" + text_type(obj.key) elif isinstance(obj, Column) and isinstance(obj.table, Table): id_ = ( "column:" + text_type(obj.table.key) + ":" + text_type(obj.key) ) elif isinstance(obj, Session): id_ = "session:" elif isinstance(obj, Engine): id_ = "engine:" else: return None return id_ pickler.persistent_id = persistent_id return pickler our_ids = re.compile( r"(mapperprop\|mapper\|table\|column\|session\|attribute\|engine):(.)" ) def Deserializer(file, metadata=None, scoped_session=None, engine=None): unpickler = pickle.Unpickler(file) def get_engine(): if engine: return engine elif scoped_session and scoped_session().bind: return scoped_session().bind elif metadata and metadata.bind: return metadata.bind else: return None def persistent_load(id_): m = our_ids.match(text_type(id_)) if not m: return None else: type_, args = m.group(1, 2) if type_ == "attribute": key, clsarg = args.split(":") cls = pickle.loads(b64decode(clsarg)) return getattr(cls, key) elif type_ == "mapper": cls = pickle.loads(b64decode(args)) return class_mapper(cls) elif type_ == "mapperprop": mapper, keyname = args.split(":") cls = pickle.loads(b64decode(mapper)) return class_mapper(cls).attrs[keyname] elif type_ == "table": return metadata.tables[args] elif type_ == "column": table, colname = args.split(":") return metadata.tables[table].c[colname] elif type_ == "session": return scoped_session() elif type_ == "engine": return get_engine() else: raise Exception("Unknown token: %s" % type_) unpickler.persistent_load = persistent_load return unpickler def dumps(obj, protocol=pickle.HIGHEST_PROTOCOL): buf = byte_buffer() pickler = Serializer(buf, protocol) pickler.dump(obj) return buf.getvalue() def loads(data, metadata=None, scoped_session=None, engine=None): buf = byte_buffer(data) unpickler = Deserializer(buf, metadata, scoped_session, engine) return unpickler.load()	5.65 KB	2026-01-20 12:58

Terminal