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/declarative/__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 .api import AbstractConcreteBase from .api import as_declarative from .api import comparable_using from .api import ConcreteBase from .api import declarative_base from .api import DeclarativeMeta from .api import declared_attr from .api import DeferredReflection from .api import has_inherited_table from .api import instrument_declarative from .api import synonym_for __all__ = [ "declarative_base", "synonym_for", "has_inherited_table", "comparable_using", "instrument_declarative", "declared_attr", "as_declarative", "ConcreteBase", "AbstractConcreteBase", "DeclarativeMeta", "DeferredReflection", ]	902 B	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 api.py # ext/declarative/api.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 """Public API functions and helpers for declarative.""" import re import weakref from .base import _add_attribute from .base import _as_declarative from .base import _declarative_constructor from .base import _DeferredMapperConfig from .base import _del_attribute from .base import _get_immediate_cls_attr from .clsregistry import _class_resolver from ... import exc from ... import inspection from ... import util from ...orm import attributes from ...orm import comparable_property from ...orm import exc as orm_exc from ...orm import interfaces from ...orm import properties from ...orm import synonym as _orm_synonym from ...orm.base import _inspect_mapped_class from ...orm.base import _mapper_or_none from ...orm.util import polymorphic_union from ...schema import MetaData from ...schema import Table from ...util import hybridmethod from ...util import hybridproperty from ...util import OrderedDict def instrument_declarative(cls, registry, metadata): """Given a class, configure the class declaratively, using the given registry, which can be any dictionary, and MetaData object. """ if "_decl_class_registry" in cls.__dict__: raise exc.InvalidRequestError( "Class %r already has been " "instrumented declaratively" % cls ) cls._decl_class_registry = registry cls.metadata = metadata _as_declarative(cls, cls.__name__, cls.__dict__) def has_inherited_table(cls): """Given a class, return True if any of the classes it inherits from has a mapped table, otherwise return False. This is used in declarative mixins to build attributes that behave differently for the base class vs. a subclass in an inheritance hierarchy. .. seealso:: :ref:`decl_mixin_inheritance` """ for class_ in cls.__mro__[1:]: if getattr(class_, "__table__", None) is not None: return True return False class DeclarativeMeta(type): def __init__(cls, classname, bases, dict_, kw): if "_decl_class_registry" not in cls.__dict__: _as_declarative(cls, classname, cls.__dict__) type.__init__(cls, classname, bases, dict_) def __setattr__(cls, key, value): _add_attribute(cls, key, value) def __delattr__(cls, key): _del_attribute(cls, key) def synonym_for(name, map_column=False): """Decorator that produces an :func:`_orm.synonym` attribute in conjunction with a Python descriptor. The function being decorated is passed to :func:`_orm.synonym` as the :paramref:`.orm.synonym.descriptor` parameter:: class MyClass(Base): __tablename__ = 'my_table' id = Column(Integer, primary_key=True) _job_status = Column("job_status", String(50)) @synonym_for("job_status") @property def job_status(self): return "Status: %s" % self._job_status The :ref:`hybrid properties <mapper_hybrids>` feature of SQLAlchemy is typically preferred instead of synonyms, which is a more legacy feature. .. seealso:: :ref:`synonyms` - Overview of synonyms :func:`_orm.synonym` - the mapper-level function :ref:`mapper_hybrids` - The Hybrid Attribute extension provides an updated approach to augmenting attribute behavior more flexibly than can be achieved with synonyms. """ def decorate(fn): return _orm_synonym(name, map_column=map_column, descriptor=fn) return decorate def comparable_using(comparator_factory): """Decorator, allow a Python @property to be used in query criteria. This is a decorator front end to :func:`~sqlalchemy.orm.comparable_property` that passes through the comparator_factory and the function being decorated:: @comparable_using(MyComparatorType) @property def prop(self): return 'special sauce' The regular ``comparable_property()`` is also usable directly in a declarative setting and may be convenient for read/write properties:: prop = comparable_property(MyComparatorType) """ def decorate(fn): return comparable_property(comparator_factory, fn) return decorate class declared_attr(interfaces._MappedAttribute, property): """Mark a class-level method as representing the definition of a mapped property or special declarative member name. @declared_attr turns the attribute into a scalar-like property that can be invoked from the uninstantiated class. Declarative treats attributes specifically marked with @declared_attr as returning a construct that is specific to mapping or declarative table configuration. The name of the attribute is that of what the non-dynamic version of the attribute would be. @declared_attr is more often than not applicable to mixins, to define relationships that are to be applied to different implementors of the class:: class ProvidesUser(object): "A mixin that adds a 'user' relationship to classes." @declared_attr def user(self): return relationship("User") It also can be applied to mapped classes, such as to provide a "polymorphic" scheme for inheritance:: class Employee(Base): id = Column(Integer, primary_key=True) type = Column(String(50), nullable=False) @declared_attr def __tablename__(cls): return cls.__name__.lower() @declared_attr def __mapper_args__(cls): if cls.__name__ == 'Employee': return { "polymorphic_on":cls.type, "polymorphic_identity":"Employee" } else: return {"polymorphic_identity":cls.__name__} """ def __init__(self, fget, cascading=False): super(declared_attr, self).__init__(fget) self.__doc__ = fget.__doc__ self._cascading = cascading def __get__(desc, self, cls): reg = cls.__dict__.get("_sa_declared_attr_reg", None) if reg is None: if ( not re.match(r"^__.+__$", desc.fget.__name__) and attributes.manager_of_class(cls) is None ): util.warn( "Unmanaged access of declarative attribute %s from " "non-mapped class %s" % (desc.fget.__name__, cls.__name__) ) return desc.fget(cls) elif desc in reg: return reg[desc] else: reg[desc] = obj = desc.fget(cls) return obj @hybridmethod def _stateful(cls, kw): return _stateful_declared_attr(*kw) @hybridproperty def cascading(cls): """Mark a :class:`.declared_attr` as cascading. This is a special-use modifier which indicates that a column or MapperProperty-based declared attribute should be configured distinctly per mapped subclass, within a mapped-inheritance scenario. .. warning:: The :attr:`.declared_attr.cascading` modifier has several limitations: The flag only applies to the use of :class:`.declared_attr` on declarative mixin classes and ``__abstract__`` classes; it currently has no effect when used on a mapped class directly. * The flag only applies to normally-named attributes, e.g. not any special underscore attributes such as ``__tablename__``. On these attributes it has no effect. * The flag currently does not allow further overrides down the class hierarchy; if a subclass tries to override the attribute, a warning is emitted and the overridden attribute is skipped. This is a limitation that it is hoped will be resolved at some point. Below, both MyClass as well as MySubClass will have a distinct ``id`` Column object established:: class HasIdMixin(object): @declared_attr.cascading def id(cls): if has_inherited_table(cls): return Column( ForeignKey('myclass.id'), primary_key=True ) else: return Column(Integer, primary_key=True) class MyClass(HasIdMixin, Base): __tablename__ = 'myclass' # ... class MySubClass(MyClass): "" # ... The behavior of the above configuration is that ``MySubClass`` will refer to both its own ``id`` column as well as that of ``MyClass`` underneath the attribute named ``some_id``. .. seealso:: :ref:`declarative_inheritance` :ref:`mixin_inheritance_columns` """ return cls._stateful(cascading=True) class _stateful_declared_attr(declared_attr): def __init__(self, kw): self.kw = kw def _stateful(self, kw): new_kw = self.kw.copy() new_kw.update(kw) return _stateful_declared_attr(new_kw) def __call__(self, fn): return declared_attr(fn, self.kw) def declarative_base( bind=None, metadata=None, mapper=None, cls=object, name="Base", constructor=_declarative_constructor, class_registry=None, metaclass=DeclarativeMeta, ): r"""Construct a base class for declarative class definitions. The new base class will be given a metaclass that produces appropriate :class:`~sqlalchemy.schema.Table` objects and makes the appropriate :func:`~sqlalchemy.orm.mapper` calls based on the information provided declaratively in the class and any subclasses of the class. :param bind: An optional :class:`~sqlalchemy.engine.Connectable`, will be assigned the ``bind`` attribute on the :class:`~sqlalchemy.schema.MetaData` instance. :param metadata: An optional :class:`~sqlalchemy.schema.MetaData` instance. All :class:`~sqlalchemy.schema.Table` objects implicitly declared by subclasses of the base will share this MetaData. A MetaData instance will be created if none is provided. The :class:`~sqlalchemy.schema.MetaData` instance will be available via the `metadata` attribute of the generated declarative base class. :param mapper: An optional callable, defaults to :func:`~sqlalchemy.orm.mapper`. Will be used to map subclasses to their Tables. :param cls: Defaults to :class:`object`. A type to use as the base for the generated declarative base class. May be a class or tuple of classes. :param name: Defaults to ``Base``. The display name for the generated class. Customizing this is not required, but can improve clarity in tracebacks and debugging. :param constructor: Defaults to :func:`~sqlalchemy.ext.declarative.base._declarative_constructor`, an __init__ implementation that assigns \kwargs for declared fields and relationships to an instance. If ``None`` is supplied, no __init__ will be provided and construction will fall back to cls.__init__ by way of the normal Python semantics. :param class_registry: optional dictionary that will serve as the registry of class names-> mapped classes when string names are used to identify classes inside of :func:`_orm.relationship` and others. Allows two or more declarative base classes to share the same registry of class names for simplified inter-base relationships. :param metaclass: Defaults to :class:`.DeclarativeMeta`. A metaclass or __metaclass__ compatible callable to use as the meta type of the generated declarative base class. .. versionchanged:: 1.1 if :paramref:`.declarative_base.cls` is a single class (rather than a tuple), the constructed base class will inherit its docstring. .. seealso:: :func:`.as_declarative` """ lcl_metadata = metadata or MetaData() if bind: lcl_metadata.bind = bind if class_registry is None: class_registry = weakref.WeakValueDictionary() bases = not isinstance(cls, tuple) and (cls,) or cls class_dict = dict( _decl_class_registry=class_registry, metadata=lcl_metadata ) if isinstance(cls, type): class_dict["__doc__"] = cls.__doc__ if constructor: class_dict["__init__"] = constructor if mapper: class_dict["__mapper_cls__"] = mapper return metaclass(name, bases, class_dict) def as_declarative(kw): """ Class decorator for :func:`.declarative_base`. Provides a syntactical shortcut to the ``cls`` argument sent to :func:`.declarative_base`, allowing the base class to be converted in-place to a "declarative" base:: from sqlalchemy.ext.declarative import as_declarative @as_declarative() class Base(object): @declared_attr def __tablename__(cls): return cls.__name__.lower() id = Column(Integer, primary_key=True) class MyMappedClass(Base): # ... All keyword arguments passed to :func:`.as_declarative` are passed along to :func:`.declarative_base`. .. seealso:: :func:`.declarative_base` """ def decorate(cls): kw["cls"] = cls kw["name"] = cls.__name__ return declarative_base(*kw) return decorate class ConcreteBase(object): """A helper class for 'concrete' declarative mappings. :class:`.ConcreteBase` will use the :func:`.polymorphic_union` function automatically, against all tables mapped as a subclass to this class. The function is called via the ``__declare_last__()`` function, which is essentially a hook for the :meth:`.after_configured` event. :class:`.ConcreteBase` produces a mapped table for the class itself. Compare to :class:`.AbstractConcreteBase`, which does not. Example:: from sqlalchemy.ext.declarative import ConcreteBase class Employee(ConcreteBase, Base): __tablename__ = 'employee' employee_id = Column(Integer, primary_key=True) name = Column(String(50)) __mapper_args__ = { 'polymorphic_identity':'employee', 'concrete':True} class Manager(Employee): __tablename__ = 'manager' employee_id = Column(Integer, primary_key=True) name = Column(String(50)) manager_data = Column(String(40)) __mapper_args__ = { 'polymorphic_identity':'manager', 'concrete':True} The name of the discriminator column used by :func:`.polymorphic_union` defaults to the name ``type``. To suit the use case of a mapping where an actual column in a mapped table is already named ``type``, the discriminator name can be configured by setting the ``_concrete_discriminator_name`` attribute:: class Employee(ConcreteBase, Base): _concrete_discriminator_name = '_concrete_discriminator' .. versionadded:: 1.3.19 Added the ``_concrete_discriminator_name`` attribute to :class:`_declarative.ConcreteBase` so that the virtual discriminator column name can be customized. .. seealso:: :class:`.AbstractConcreteBase` :ref:`concrete_inheritance` """ @classmethod def _create_polymorphic_union(cls, mappers, discriminator_name): return polymorphic_union( OrderedDict( (mp.polymorphic_identity, mp.local_table) for mp in mappers ), discriminator_name, "pjoin", ) @classmethod def __declare_first__(cls): m = cls.__mapper__ if m.with_polymorphic: return discriminator_name = ( _get_immediate_cls_attr(cls, "_concrete_discriminator_name") or "type" ) mappers = list(m.self_and_descendants) pjoin = cls._create_polymorphic_union(mappers, discriminator_name) m._set_with_polymorphic(("", pjoin)) m._set_polymorphic_on(pjoin.c[discriminator_name]) class AbstractConcreteBase(ConcreteBase): """A helper class for 'concrete' declarative mappings. :class:`.AbstractConcreteBase` will use the :func:`.polymorphic_union` function automatically, against all tables mapped as a subclass to this class. The function is called via the ``__declare_last__()`` function, which is essentially a hook for the :meth:`.after_configured` event. :class:`.AbstractConcreteBase` does produce a mapped class for the base class, however it is not persisted to any table; it is instead mapped directly to the "polymorphic" selectable directly and is only used for selecting. Compare to :class:`.ConcreteBase`, which does create a persisted table for the base class. .. note:: The :class:`.AbstractConcreteBase` class does not intend to set up the mapping for the base class until all the subclasses have been defined, as it needs to create a mapping against a selectable that will include all subclass tables. In order to achieve this, it waits for the mapper configuration event to occur, at which point it scans through all the configured subclasses and sets up a mapping that will query against all subclasses at once. While this event is normally invoked automatically, in the case of :class:`.AbstractConcreteBase`, it may be necessary to invoke it explicitly after all subclass mappings are defined, if the first operation is to be a query against this base class. To do so, invoke :func:`.configure_mappers` once all the desired classes have been configured:: from sqlalchemy.orm import configure_mappers configure_mappers() .. seealso:: :func:`_orm.configure_mappers` Example:: from sqlalchemy.ext.declarative import AbstractConcreteBase class Employee(AbstractConcreteBase, Base): pass class Manager(Employee): __tablename__ = 'manager' employee_id = Column(Integer, primary_key=True) name = Column(String(50)) manager_data = Column(String(40)) __mapper_args__ = { 'polymorphic_identity':'manager', 'concrete':True} configure_mappers() The abstract base class is handled by declarative in a special way; at class configuration time, it behaves like a declarative mixin or an ``__abstract__`` base class. Once classes are configured and mappings are produced, it then gets mapped itself, but after all of its descendants. This is a very unique system of mapping not found in any other SQLAlchemy system. Using this approach, we can specify columns and properties that will take place on mapped subclasses, in the way that we normally do as in :ref:`declarative_mixins`:: class Company(Base): __tablename__ = 'company' id = Column(Integer, primary_key=True) class Employee(AbstractConcreteBase, Base): employee_id = Column(Integer, primary_key=True) @declared_attr def company_id(cls): return Column(ForeignKey('company.id')) @declared_attr def company(cls): return relationship("Company") class Manager(Employee): __tablename__ = 'manager' name = Column(String(50)) manager_data = Column(String(40)) __mapper_args__ = { 'polymorphic_identity':'manager', 'concrete':True} configure_mappers() When we make use of our mappings however, both ``Manager`` and ``Employee`` will have an independently usable ``.company`` attribute:: session.query(Employee).filter(Employee.company.has(id=5)) .. versionchanged:: 1.0.0 - The mechanics of :class:`.AbstractConcreteBase` have been reworked to support relationships established directly on the abstract base, without any special configurational steps. .. seealso:: :class:`.ConcreteBase` :ref:`concrete_inheritance` """ __no_table__ = True @classmethod def __declare_first__(cls): cls._sa_decl_prepare_nocascade() @classmethod def _sa_decl_prepare_nocascade(cls): if getattr(cls, "__mapper__", None): return to_map = _DeferredMapperConfig.config_for_cls(cls) # can't rely on 'self_and_descendants' here # since technically an immediate subclass # might not be mapped, but a subclass # may be. mappers = [] stack = list(cls.__subclasses__()) while stack: klass = stack.pop() stack.extend(klass.__subclasses__()) mn = _mapper_or_none(klass) if mn is not None: mappers.append(mn) discriminator_name = ( _get_immediate_cls_attr(cls, "_concrete_discriminator_name") or "type" ) pjoin = cls._create_polymorphic_union(mappers, discriminator_name) # For columns that were declared on the class, these # are normally ignored with the "__no_table__" mapping, # unless they have a different attribute key vs. col name # and are in the properties argument. # In that case, ensure we update the properties entry # to the correct column from the pjoin target table. declared_cols = set(to_map.declared_columns) for k, v in list(to_map.properties.items()): if v in declared_cols: to_map.properties[k] = pjoin.c[v.key] to_map.local_table = pjoin m_args = to_map.mapper_args_fn or dict def mapper_args(): args = m_args() args["polymorphic_on"] = pjoin.c[discriminator_name] return args to_map.mapper_args_fn = mapper_args m = to_map.map() for scls in cls.__subclasses__(): sm = _mapper_or_none(scls) if sm and sm.concrete and cls in scls.__bases__: sm._set_concrete_base(m) @classmethod def _sa_raise_deferred_config(cls): raise orm_exc.UnmappedClassError( cls, msg="Class %s is a subclass of AbstractConcreteBase and " "has a mapping pending until all subclasses are defined. " "Call the sqlalchemy.orm.configure_mappers() function after " "all subclasses have been defined to " "complete the mapping of this class." % orm_exc._safe_cls_name(cls), ) class DeferredReflection(object): """A helper class for construction of mappings based on a deferred reflection step. Normally, declarative can be used with reflection by setting a :class:`_schema.Table` object using autoload=True as the ``__table__`` attribute on a declarative class. The caveat is that the :class:`_schema.Table` must be fully reflected, or at the very least have a primary key column, at the point at which a normal declarative mapping is constructed, meaning the :class:`_engine.Engine` must be available at class declaration time. The :class:`.DeferredReflection` mixin moves the construction of mappers to be at a later point, after a specific method is called which first reflects all :class:`_schema.Table` objects created so far. Classes can define it as such:: from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.ext.declarative import DeferredReflection Base = declarative_base() class MyClass(DeferredReflection, Base): __tablename__ = 'mytable' Above, ``MyClass`` is not yet mapped. After a series of classes have been defined in the above fashion, all tables can be reflected and mappings created using :meth:`.prepare`:: engine = create_engine("someengine://...") DeferredReflection.prepare(engine) The :class:`.DeferredReflection` mixin can be applied to individual classes, used as the base for the declarative base itself, or used in a custom abstract class. Using an abstract base allows that only a subset of classes to be prepared for a particular prepare step, which is necessary for applications that use more than one engine. For example, if an application has two engines, you might use two bases, and prepare each separately, e.g.:: class ReflectedOne(DeferredReflection, Base): __abstract__ = True class ReflectedTwo(DeferredReflection, Base): __abstract__ = True class MyClass(ReflectedOne): __tablename__ = 'mytable' class MyOtherClass(ReflectedOne): __tablename__ = 'myothertable' class YetAnotherClass(ReflectedTwo): __tablename__ = 'yetanothertable' # ... etc. Above, the class hierarchies for ``ReflectedOne`` and ``ReflectedTwo`` can be configured separately:: ReflectedOne.prepare(engine_one) ReflectedTwo.prepare(engine_two) """ @classmethod def prepare(cls, engine): """Reflect all :class:`_schema.Table` objects for all current :class:`.DeferredReflection` subclasses""" to_map = _DeferredMapperConfig.classes_for_base(cls) for thingy in to_map: cls._sa_decl_prepare(thingy.local_table, engine) thingy.map() mapper = thingy.cls.__mapper__ metadata = mapper.class_.metadata for rel in mapper._props.values(): if ( isinstance(rel, properties.RelationshipProperty) and rel.secondary is not None ): if isinstance(rel.secondary, Table): cls._reflect_table(rel.secondary, engine) elif isinstance(rel.secondary, _class_resolver): rel.secondary._resolvers += ( cls._sa_deferred_table_resolver(engine, metadata), ) @classmethod def _sa_deferred_table_resolver(cls, engine, metadata): def _resolve(key): t1 = Table(key, metadata) cls._reflect_table(t1, engine) return t1 return _resolve @classmethod def _sa_decl_prepare(cls, local_table, engine): # autoload Table, which is already # present in the metadata. This # will fill in db-loaded columns # into the existing Table object. if local_table is not None: cls._reflect_table(local_table, engine) @classmethod def _sa_raise_deferred_config(cls): raise orm_exc.UnmappedClassError( cls, msg="Class %s is a subclass of DeferredReflection. " "Mappings are not produced until the .prepare() " "method is called on the class hierarchy." % orm_exc._safe_cls_name(cls), ) @classmethod def _reflect_table(cls, table, engine): Table( table.name, table.metadata, extend_existing=True, autoload_replace=False, autoload=True, autoload_with=engine, schema=table.schema, ) @inspection._inspects(DeclarativeMeta) def _inspect_decl_meta(cls): mp = _inspect_mapped_class(cls) if mp is None: if _DeferredMapperConfig.has_cls(cls): _DeferredMapperConfig.raise_unmapped_for_cls(cls) raise orm_exc.UnmappedClassError( cls, msg="Class %s has a deferred mapping on it. It is not yet " "usable as a mapped class." % orm_exc._safe_cls_name(cls), ) return mp	28.05 KB	2026-01-20 12:58
📄 base.py # ext/declarative/base.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 """Internal implementation for declarative.""" import collections import weakref from sqlalchemy.orm import instrumentation from . import clsregistry from ... import event from ... import exc from ... import util from ...orm import class_mapper from ...orm import exc as orm_exc from ...orm import mapper from ...orm import mapperlib from ...orm import synonym from ...orm.attributes import QueryableAttribute from ...orm.base import _is_mapped_class from ...orm.base import InspectionAttr from ...orm.interfaces import MapperProperty from ...orm.properties import ColumnProperty from ...orm.properties import CompositeProperty from ...schema import Column from ...schema import Table from ...sql import expression from ...util import topological declared_attr = declarative_props = None def _declared_mapping_info(cls): # deferred mapping if _DeferredMapperConfig.has_cls(cls): return _DeferredMapperConfig.config_for_cls(cls) # regular mapping elif _is_mapped_class(cls): return class_mapper(cls, configure=False) else: return None def _resolve_for_abstract_or_classical(cls): if cls is object: return None if _get_immediate_cls_attr(cls, "__abstract__", strict=True): for sup in cls.__bases__: sup = _resolve_for_abstract_or_classical(sup) if sup is not None: return sup else: return None else: classical = _dive_for_classically_mapped_class(cls) if classical is not None: return classical else: return cls def _dive_for_classically_mapped_class(cls): # support issue #4321 # if we are within a base hierarchy, don't # search at all for classical mappings if hasattr(cls, "_decl_class_registry"): return None manager = instrumentation.manager_of_class(cls) if manager is not None: return cls else: for sup in cls.__bases__: mapper = _dive_for_classically_mapped_class(sup) if mapper is not None: return sup else: return None def _get_immediate_cls_attr(cls, attrname, strict=False): """return an attribute of the class that is either present directly on the class, e.g. not on a superclass, or is from a superclass but this superclass is a non-mapped mixin, that is, not a descendant of the declarative base and is also not classically mapped. This is used to detect attributes that indicate something about a mapped class independently from any mapped classes that it may inherit from. """ if not issubclass(cls, object): return None for base in cls.__mro__: _is_declarative_inherits = hasattr(base, "_decl_class_registry") _is_classicial_inherits = ( not _is_declarative_inherits and _dive_for_classically_mapped_class(base) is not None ) if attrname in base.__dict__ and ( base is cls or ( (base in cls.__bases__ if strict else True) and not _is_declarative_inherits and not _is_classicial_inherits ) ): return getattr(base, attrname) else: return None def _as_declarative(cls, classname, dict_): global declared_attr, declarative_props if declared_attr is None: from .api import declared_attr declarative_props = (declared_attr, util.classproperty) if _get_immediate_cls_attr(cls, "__abstract__", strict=True): return _MapperConfig.setup_mapping(cls, classname, dict_) def _check_declared_props_nocascade(obj, name, cls): if isinstance(obj, declarative_props): if getattr(obj, "_cascading", False): util.warn( "@declared_attr.cascading is not supported on the %s " "attribute on class %s. This attribute invokes for " "subclasses in any case." % (name, cls) ) return True else: return False class _MapperConfig(object): @classmethod def setup_mapping(cls, cls_, classname, dict_): defer_map = _get_immediate_cls_attr( cls_, "_sa_decl_prepare_nocascade", strict=True ) or hasattr(cls_, "_sa_decl_prepare") if defer_map: cfg_cls = _DeferredMapperConfig else: cfg_cls = _MapperConfig cfg_cls(cls_, classname, dict_) def __init__(self, cls_, classname, dict_): self.cls = cls_ # dict_ will be a dictproxy, which we can't write to, and we need to! self.dict_ = dict(dict_) self.classname = classname self.persist_selectable = None self.properties = util.OrderedDict() self.declared_columns = set() self.column_copies = {} self._setup_declared_events() # temporary registry. While early 1.0 versions # set up the ClassManager here, by API contract # we can't do that until there's a mapper. self.cls._sa_declared_attr_reg = {} self._scan_attributes() mapperlib._CONFIGURE_MUTEX.acquire() try: clsregistry.add_class(self.classname, self.cls) self._extract_mappable_attributes() self._extract_declared_columns() self._setup_table() self._setup_inheritance() self._early_mapping() finally: mapperlib._CONFIGURE_MUTEX.release() def _early_mapping(self): self.map() def _setup_declared_events(self): if _get_immediate_cls_attr(self.cls, "__declare_last__"): @event.listens_for(mapper, "after_configured") def after_configured(): self.cls.__declare_last__() if _get_immediate_cls_attr(self.cls, "__declare_first__"): @event.listens_for(mapper, "before_configured") def before_configured(): self.cls.__declare_first__() def _scan_attributes(self): cls = self.cls dict_ = self.dict_ column_copies = self.column_copies mapper_args_fn = None table_args = inherited_table_args = None tablename = None for base in cls.__mro__: class_mapped = ( base is not cls and _declared_mapping_info(base) is not None and not _get_immediate_cls_attr( base, "_sa_decl_prepare_nocascade", strict=True ) ) if not class_mapped and base is not cls: self._produce_column_copies(base) for name, obj in vars(base).items(): if name == "__mapper_args__": check_decl = _check_declared_props_nocascade( obj, name, cls ) if not mapper_args_fn and (not class_mapped or check_decl): # don't even invoke __mapper_args__ until # after we've determined everything about the # mapped table. # make a copy of it so a class-level dictionary # is not overwritten when we update column-based # arguments. def mapper_args_fn(): return dict(cls.__mapper_args__) elif name == "__tablename__": check_decl = _check_declared_props_nocascade( obj, name, cls ) if not tablename and (not class_mapped or check_decl): tablename = cls.__tablename__ elif name == "__table_args__": check_decl = _check_declared_props_nocascade( obj, name, cls ) if not table_args and (not class_mapped or check_decl): table_args = cls.__table_args__ if not isinstance( table_args, (tuple, dict, type(None)) ): raise exc.ArgumentError( "__table_args__ value must be a tuple, " "dict, or None" ) if base is not cls: inherited_table_args = True elif class_mapped: if isinstance(obj, declarative_props): util.warn( "Regular (i.e. not __special__) " "attribute '%s.%s' uses @declared_attr, " "but owning class %s is mapped - " "not applying to subclass %s." % (base.__name__, name, base, cls) ) continue elif base is not cls: # we're a mixin, abstract base, or something that is # acting like that for now. if isinstance(obj, Column): # already copied columns to the mapped class. continue elif isinstance(obj, MapperProperty): raise exc.InvalidRequestError( "Mapper properties (i.e. deferred," "column_property(), relationship(), etc.) must " "be declared as @declared_attr callables " "on declarative mixin classes." ) elif isinstance(obj, declarative_props): oldclassprop = isinstance(obj, util.classproperty) if not oldclassprop and obj._cascading: if name in dict_: # unfortunately, while we can use the user- # defined attribute here to allow a clean # override, if there's another # subclass below then it still tries to use # this. not sure if there is enough # information here to add this as a feature # later on. util.warn( "Attribute '%s' on class %s cannot be " "processed due to " "@declared_attr.cascading; " "skipping" % (name, cls) ) dict_[name] = column_copies[ obj ] = ret = obj.__get__(obj, cls) setattr(cls, name, ret) else: if oldclassprop: util.warn_deprecated( "Use of sqlalchemy.util.classproperty on " "declarative classes is deprecated." ) # access attribute using normal class access ret = getattr(cls, name) # correct for proxies created from hybrid_property # or similar. note there is no known case that # produces nested proxies, so we are only # looking one level deep right now. if ( isinstance(ret, InspectionAttr) and ret._is_internal_proxy and not isinstance( ret.original_property, MapperProperty ) ): ret = ret.descriptor dict_[name] = column_copies[obj] = ret if ( isinstance(ret, (Column, MapperProperty)) and ret.doc is None ): ret.doc = obj.__doc__ # here, the attribute is some other kind of property that # we assume is not part of the declarative mapping. # however, check for some more common mistakes else: self._warn_for_decl_attributes(base, name, obj) if inherited_table_args and not tablename: table_args = None self.table_args = table_args self.tablename = tablename self.mapper_args_fn = mapper_args_fn def _warn_for_decl_attributes(self, cls, key, c): if isinstance(c, expression.ColumnClause): util.warn( "Attribute '%s' on class %s appears to be a non-schema " "'sqlalchemy.sql.column()' " "object; this won't be part of the declarative mapping" % (key, cls) ) def _produce_column_copies(self, base): cls = self.cls dict_ = self.dict_ column_copies = self.column_copies # copy mixin columns to the mapped class for name, obj in vars(base).items(): if isinstance(obj, Column): if getattr(cls, name) is not obj: # if column has been overridden # (like by the InstrumentedAttribute of the # superclass), skip continue elif obj.foreign_keys: raise exc.InvalidRequestError( "Columns with foreign keys to other columns " "must be declared as @declared_attr callables " "on declarative mixin classes. " ) elif name not in dict_ and not ( "__table__" in dict_ and (obj.name or name) in dict_["__table__"].c ): column_copies[obj] = copy_ = obj.copy() copy_._creation_order = obj._creation_order setattr(cls, name, copy_) dict_[name] = copy_ def _extract_mappable_attributes(self): cls = self.cls dict_ = self.dict_ our_stuff = self.properties late_mapped = _get_immediate_cls_attr( cls, "_sa_decl_prepare_nocascade", strict=True ) for k in list(dict_): if k in ("__table__", "__tablename__", "__mapper_args__"): continue value = dict_[k] if isinstance(value, declarative_props): if isinstance(value, declared_attr) and value._cascading: util.warn( "Use of @declared_attr.cascading only applies to " "Declarative 'mixin' and 'abstract' classes. " "Currently, this flag is ignored on mapped class " "%s" % self.cls ) value = getattr(cls, k) elif ( isinstance(value, QueryableAttribute) and value.class_ is not cls and value.key != k ): # detect a QueryableAttribute that's already mapped being # assigned elsewhere in userland, turn into a synonym() value = synonym(value.key) setattr(cls, k, value) if ( isinstance(value, tuple) and len(value) == 1 and isinstance(value[0], (Column, MapperProperty)) ): util.warn( "Ignoring declarative-like tuple value of attribute " "'%s': possibly a copy-and-paste error with a comma " "accidentally placed at the end of the line?" % k ) continue elif not isinstance(value, (Column, MapperProperty)): # using @declared_attr for some object that # isn't Column/MapperProperty; remove from the dict_ # and place the evaluated value onto the class. if not k.startswith("__"): dict_.pop(k) self._warn_for_decl_attributes(cls, k, value) if not late_mapped: setattr(cls, k, value) continue # we expect to see the name 'metadata' in some valid cases; # however at this point we see it's assigned to something trying # to be mapped, so raise for that. elif k == "metadata": raise exc.InvalidRequestError( "Attribute name 'metadata' is reserved " "for the MetaData instance when using a " "declarative base class." ) prop = clsregistry._deferred_relationship(cls, value) our_stuff[k] = prop def _extract_declared_columns(self): our_stuff = self.properties # set up attributes in the order they were created our_stuff.sort(key=lambda key: our_stuff[key]._creation_order) # extract columns from the class dict declared_columns = self.declared_columns name_to_prop_key = collections.defaultdict(set) for key, c in list(our_stuff.items()): if isinstance(c, (ColumnProperty, CompositeProperty)): for col in c.columns: if isinstance(col, Column) and col.table is None: _undefer_column_name(key, col) if not isinstance(c, CompositeProperty): name_to_prop_key[col.name].add(key) declared_columns.add(col) elif isinstance(c, Column): _undefer_column_name(key, c) name_to_prop_key[c.name].add(key) declared_columns.add(c) # if the column is the same name as the key, # remove it from the explicit properties dict. # the normal rules for assigning column-based properties # will take over, including precedence of columns # in multi-column ColumnProperties. if key == c.key: del our_stuff[key] for name, keys in name_to_prop_key.items(): if len(keys) > 1: util.warn( "On class %r, Column object %r named " "directly multiple times, " "only one will be used: %s. " "Consider using orm.synonym instead" % (self.classname, name, (", ".join(sorted(keys)))) ) def _setup_table(self): cls = self.cls tablename = self.tablename table_args = self.table_args dict_ = self.dict_ declared_columns = self.declared_columns declared_columns = self.declared_columns = sorted( declared_columns, key=lambda c: c._creation_order ) table = None if hasattr(cls, "__table_cls__"): table_cls = util.unbound_method_to_callable(cls.__table_cls__) else: table_cls = Table if "__table__" not in dict_: if tablename is not None: args, table_kw = (), {} if table_args: if isinstance(table_args, dict): table_kw = table_args elif isinstance(table_args, tuple): if isinstance(table_args[-1], dict): args, table_kw = table_args[0:-1], table_args[-1] else: args = table_args autoload = dict_.get("__autoload__") if autoload: table_kw["autoload"] = True cls.__table__ = table = table_cls( tablename, cls.metadata, (tuple(declared_columns) + tuple(args)), table_kw ) else: table = cls.__table__ if declared_columns: for c in declared_columns: if not table.c.contains_column(c): raise exc.ArgumentError( "Can't add additional column %r when " "specifying __table__" % c.key ) self.local_table = table def _setup_inheritance(self): table = self.local_table cls = self.cls table_args = self.table_args declared_columns = self.declared_columns # since we search for classical mappings now, search for # multiple mapped bases as well and raise an error. inherits = [] for c in cls.__bases__: c = _resolve_for_abstract_or_classical(c) if c is None: continue if _declared_mapping_info( c ) is not None and not _get_immediate_cls_attr( c, "_sa_decl_prepare_nocascade", strict=True ): inherits.append(c) if inherits: if len(inherits) > 1: raise exc.InvalidRequestError( "Class %s has multiple mapped bases: %r" % (cls, inherits) ) self.inherits = inherits[0] else: self.inherits = None if ( table is None and self.inherits is None and not _get_immediate_cls_attr(cls, "__no_table__") ): raise exc.InvalidRequestError( "Class %r does not have a __table__ or __tablename__ " "specified and does not inherit from an existing " "table-mapped class." % cls ) elif self.inherits: inherited_mapper = _declared_mapping_info(self.inherits) inherited_table = inherited_mapper.local_table inherited_persist_selectable = inherited_mapper.persist_selectable if table is None: # single table inheritance. # ensure no table args if table_args: raise exc.ArgumentError( "Can't place __table_args__ on an inherited class " "with no table." ) # add any columns declared here to the inherited table. for c in declared_columns: if c.name in inherited_table.c: if inherited_table.c[c.name] is c: continue raise exc.ArgumentError( "Column '%s' on class %s conflicts with " "existing column '%s'" % (c, cls, inherited_table.c[c.name]) ) if c.primary_key: raise exc.ArgumentError( "Can't place primary key columns on an inherited " "class with no table." ) inherited_table.append_column(c) if ( inherited_persist_selectable is not None and inherited_persist_selectable is not inherited_table ): inherited_persist_selectable._refresh_for_new_column(c) def _prepare_mapper_arguments(self): properties = self.properties if self.mapper_args_fn: mapper_args = self.mapper_args_fn() else: mapper_args = {} # make sure that column copies are used rather # than the original columns from any mixins for k in ("version_id_col", "polymorphic_on"): if k in mapper_args: v = mapper_args[k] mapper_args[k] = self.column_copies.get(v, v) assert ( "inherits" not in mapper_args ), "Can't specify 'inherits' explicitly with declarative mappings" if self.inherits: mapper_args["inherits"] = self.inherits if self.inherits and not mapper_args.get("concrete", False): # single or joined inheritance # exclude any cols on the inherited table which are # not mapped on the parent class, to avoid # mapping columns specific to sibling/nephew classes inherited_mapper = _declared_mapping_info(self.inherits) inherited_table = inherited_mapper.local_table if "exclude_properties" not in mapper_args: mapper_args["exclude_properties"] = exclude_properties = set( [ c.key for c in inherited_table.c if c not in inherited_mapper._columntoproperty ] ).union(inherited_mapper.exclude_properties or ()) exclude_properties.difference_update( [c.key for c in self.declared_columns] ) # look through columns in the current mapper that # are keyed to a propname different than the colname # (if names were the same, we'd have popped it out above, # in which case the mapper makes this combination). # See if the superclass has a similar column property. # If so, join them together. for k, col in list(properties.items()): if not isinstance(col, expression.ColumnElement): continue if k in inherited_mapper._props: p = inherited_mapper._props[k] if isinstance(p, ColumnProperty): # note here we place the subclass column # first. See [ticket:1892] for background. properties[k] = [col] + p.columns result_mapper_args = mapper_args.copy() result_mapper_args["properties"] = properties self.mapper_args = result_mapper_args def map(self): self._prepare_mapper_arguments() if hasattr(self.cls, "__mapper_cls__"): mapper_cls = util.unbound_method_to_callable( self.cls.__mapper_cls__ ) else: mapper_cls = mapper self.cls.__mapper__ = mp_ = mapper_cls( self.cls, self.local_table, self.mapper_args ) del self.cls._sa_declared_attr_reg return mp_ class _DeferredMapperConfig(_MapperConfig): _configs = util.OrderedDict() def _early_mapping(self): pass @property def cls(self): return self._cls() @cls.setter def cls(self, class_): self._cls = weakref.ref(class_, self._remove_config_cls) self._configs[self._cls] = self @classmethod def _remove_config_cls(cls, ref): cls._configs.pop(ref, None) @classmethod def has_cls(cls, class_): # 2.6 fails on weakref if class_ is an old style class return isinstance(class_, type) and weakref.ref(class_) in cls._configs @classmethod def raise_unmapped_for_cls(cls, class_): if hasattr(class_, "_sa_raise_deferred_config"): class_._sa_raise_deferred_config() raise orm_exc.UnmappedClassError( class_, msg="Class %s has a deferred mapping on it. It is not yet " "usable as a mapped class." % orm_exc._safe_cls_name(class_), ) @classmethod def config_for_cls(cls, class_): return cls._configs[weakref.ref(class_)] @classmethod def classes_for_base(cls, base_cls, sort=True): classes_for_base = [ m for m, cls_ in [(m, m.cls) for m in cls._configs.values()] if cls_ is not None and issubclass(cls_, base_cls) ] if not sort: return classes_for_base all_m_by_cls = dict((m.cls, m) for m in classes_for_base) tuples = [] for m_cls in all_m_by_cls: tuples.extend( (all_m_by_cls[base_cls], all_m_by_cls[m_cls]) for base_cls in m_cls.__bases__ if base_cls in all_m_by_cls ) return list(topological.sort(tuples, classes_for_base)) def map(self): self._configs.pop(self._cls, None) return super(_DeferredMapperConfig, self).map() def _add_attribute(cls, key, value): """add an attribute to an existing declarative class. This runs through the logic to determine MapperProperty, adds it to the Mapper, adds a column to the mapped Table, etc. """ if "__mapper__" in cls.__dict__: if isinstance(value, Column): _undefer_column_name(key, value) cls.__table__.append_column(value) cls.__mapper__.add_property(key, value) elif isinstance(value, ColumnProperty): for col in value.columns: if isinstance(col, Column) and col.table is None: _undefer_column_name(key, col) cls.__table__.append_column(col) cls.__mapper__.add_property(key, value) elif isinstance(value, MapperProperty): cls.__mapper__.add_property( key, clsregistry._deferred_relationship(cls, value) ) elif isinstance(value, QueryableAttribute) and value.key != key: # detect a QueryableAttribute that's already mapped being # assigned elsewhere in userland, turn into a synonym() value = synonym(value.key) cls.__mapper__.add_property( key, clsregistry._deferred_relationship(cls, value) ) else: type.__setattr__(cls, key, value) cls.__mapper__._expire_memoizations() else: type.__setattr__(cls, key, value) def _del_attribute(cls, key): if ( "__mapper__" in cls.__dict__ and key in cls.__dict__ and not cls.__mapper__._dispose_called ): value = cls.__dict__[key] if isinstance( value, (Column, ColumnProperty, MapperProperty, QueryableAttribute) ): raise NotImplementedError( "Can't un-map individual mapped attributes on a mapped class." ) else: type.__delattr__(cls, key) cls.__mapper__._expire_memoizations() else: type.__delattr__(cls, key) def _declarative_constructor(self, *kwargs): """A simple constructor that allows initialization from kwargs. Sets attributes on the constructed instance using the names and values in ``kwargs``. Only keys that are present as attributes of the instance's class are allowed. These could be, for example, any mapped columns or relationships. """ cls_ = type(self) for k in kwargs: if not hasattr(cls_, k): raise TypeError( "%r is an invalid keyword argument for %s" % (k, cls_.__name__) ) setattr(self, k, kwargs[k]) _declarative_constructor.__name__ = "__init__" def _undefer_column_name(key, column): if column.key is None: column.key = key if column.name is None: column.name = key	31.34 KB	2026-01-20 12:58
📄 clsregistry.py # ext/declarative/clsregistry.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 """Routines to handle the string class registry used by declarative. This system allows specification of classes and expressions used in :func:`_orm.relationship` using strings. """ import weakref from ... import exc from ... import inspection from ... import util from ...orm import class_mapper from ...orm import interfaces from ...orm.properties import ColumnProperty from ...orm.properties import RelationshipProperty from ...orm.properties import SynonymProperty from ...schema import _get_table_key # strong references to registries which we place in # the _decl_class_registry, which is usually weak referencing. # the internal registries here link to classes with weakrefs and remove # themselves when all references to contained classes are removed. _registries = set() def add_class(classname, cls): """Add a class to the _decl_class_registry associated with the given declarative class. """ if classname in cls._decl_class_registry: # class already exists. existing = cls._decl_class_registry[classname] if not isinstance(existing, _MultipleClassMarker): existing = cls._decl_class_registry[ classname ] = _MultipleClassMarker([cls, existing]) else: cls._decl_class_registry[classname] = cls try: root_module = cls._decl_class_registry["_sa_module_registry"] except KeyError: cls._decl_class_registry[ "_sa_module_registry" ] = root_module = _ModuleMarker("_sa_module_registry", None) tokens = cls.__module__.split(".") # build up a tree like this: # modulename: myapp.snacks.nuts # # myapp->snack->nuts->(classes) # snack->nuts->(classes) # nuts->(classes) # # this allows partial token paths to be used. while tokens: token = tokens.pop(0) module = root_module.get_module(token) for token in tokens: module = module.get_module(token) module.add_class(classname, cls) class _MultipleClassMarker(object): """refers to multiple classes of the same name within _decl_class_registry. """ __slots__ = "on_remove", "contents", "__weakref__" def __init__(self, classes, on_remove=None): self.on_remove = on_remove self.contents = set( [weakref.ref(item, self._remove_item) for item in classes] ) _registries.add(self) def __iter__(self): return (ref() for ref in self.contents) def attempt_get(self, path, key): if len(self.contents) > 1: raise exc.InvalidRequestError( 'Multiple classes found for path "%s" ' "in the registry of this declarative " "base. Please use a fully module-qualified path." % (".".join(path + [key])) ) else: ref = list(self.contents)[0] cls = ref() if cls is None: raise NameError(key) return cls def _remove_item(self, ref): self.contents.remove(ref) if not self.contents: _registries.discard(self) if self.on_remove: self.on_remove() def add_item(self, item): # protect against class registration race condition against # asynchronous garbage collection calling _remove_item, # [ticket:3208] modules = set( [ cls.__module__ for cls in [ref() for ref in self.contents] if cls is not None ] ) if item.__module__ in modules: util.warn( "This declarative base already contains a class with the " "same class name and module name as %s.%s, and will " "be replaced in the string-lookup table." % (item.__module__, item.__name__) ) self.contents.add(weakref.ref(item, self._remove_item)) class _ModuleMarker(object): """Refers to a module name within _decl_class_registry. """ __slots__ = "parent", "name", "contents", "mod_ns", "path", "__weakref__" def __init__(self, name, parent): self.parent = parent self.name = name self.contents = {} self.mod_ns = _ModNS(self) if self.parent: self.path = self.parent.path + [self.name] else: self.path = [] _registries.add(self) def __contains__(self, name): return name in self.contents def __getitem__(self, name): return self.contents[name] def _remove_item(self, name): self.contents.pop(name, None) if not self.contents and self.parent is not None: self.parent._remove_item(self.name) _registries.discard(self) def resolve_attr(self, key): return getattr(self.mod_ns, key) def get_module(self, name): if name not in self.contents: marker = _ModuleMarker(name, self) self.contents[name] = marker else: marker = self.contents[name] return marker def add_class(self, name, cls): if name in self.contents: existing = self.contents[name] existing.add_item(cls) else: existing = self.contents[name] = _MultipleClassMarker( [cls], on_remove=lambda: self._remove_item(name) ) class _ModNS(object): __slots__ = ("__parent",) def __init__(self, parent): self.__parent = parent def __getattr__(self, key): try: value = self.__parent.contents[key] except KeyError: pass else: if value is not None: if isinstance(value, _ModuleMarker): return value.mod_ns else: assert isinstance(value, _MultipleClassMarker) return value.attempt_get(self.__parent.path, key) raise AttributeError( "Module %r has no mapped classes " "registered under the name %r" % (self.__parent.name, key) ) class _GetColumns(object): __slots__ = ("cls",) def __init__(self, cls): self.cls = cls def __getattr__(self, key): mp = class_mapper(self.cls, configure=False) if mp: if key not in mp.all_orm_descriptors: raise exc.InvalidRequestError( "Class %r does not have a mapped column named %r" % (self.cls, key) ) desc = mp.all_orm_descriptors[key] if desc.extension_type is interfaces.NOT_EXTENSION: prop = desc.property if isinstance(prop, SynonymProperty): key = prop.name elif not isinstance(prop, ColumnProperty): raise exc.InvalidRequestError( "Property %r is not an instance of" " ColumnProperty (i.e. does not correspond" " directly to a Column)." % key ) return getattr(self.cls, key) inspection._inspects(_GetColumns)( lambda target: inspection.inspect(target.cls) ) class _GetTable(object): __slots__ = "key", "metadata" def __init__(self, key, metadata): self.key = key self.metadata = metadata def __getattr__(self, key): return self.metadata.tables[_get_table_key(key, self.key)] def _determine_container(key, value): if isinstance(value, _MultipleClassMarker): value = value.attempt_get([], key) return _GetColumns(value) class _class_resolver(object): def __init__(self, cls, prop, fallback, arg, favor_tables=False): self.cls = cls self.prop = prop self.arg = self._declarative_arg = arg self.fallback = fallback self._dict = util.PopulateDict(self._access_cls) self._resolvers = () self.favor_tables = favor_tables def _access_cls(self, key): cls = self.cls if self.favor_tables: if key in cls.metadata.tables: return cls.metadata.tables[key] elif key in cls.metadata._schemas: return _GetTable(key, cls.metadata) if key in cls._decl_class_registry: return _determine_container(key, cls._decl_class_registry[key]) if not self.favor_tables: if key in cls.metadata.tables: return cls.metadata.tables[key] elif key in cls.metadata._schemas: return _GetTable(key, cls.metadata) if ( "_sa_module_registry" in cls._decl_class_registry and key in cls._decl_class_registry["_sa_module_registry"] ): registry = cls._decl_class_registry["_sa_module_registry"] return registry.resolve_attr(key) elif self._resolvers: for resolv in self._resolvers: value = resolv(key) if value is not None: return value return self.fallback[key] def _raise_for_name(self, name, err): util.raise_( exc.InvalidRequestError( "When initializing mapper %s, expression %r failed to " "locate a name (%r). If this is a class name, consider " "adding this relationship() to the %r class after " "both dependent classes have been defined." % (self.prop.parent, self.arg, name, self.cls) ), from_=err, ) def _resolve_name(self): name = self.arg d = self._dict rval = None try: for token in name.split("."): if rval is None: rval = d[token] else: rval = getattr(rval, token) except KeyError as err: self._raise_for_name(name, err) except NameError as n: self._raise_for_name(n.args[0], n) else: if isinstance(rval, _GetColumns): return rval.cls else: return rval def __call__(self): try: x = eval(self.arg, globals(), self._dict) if isinstance(x, _GetColumns): return x.cls else: return x except NameError as n: self._raise_for_name(n.args[0], n) def _resolver(cls, prop): import sqlalchemy from sqlalchemy.orm import foreign, remote fallback = sqlalchemy.__dict__.copy() fallback.update({"foreign": foreign, "remote": remote}) def resolve_arg(arg, favor_tables=False): return _class_resolver( cls, prop, fallback, arg, favor_tables=favor_tables ) def resolve_name(arg): return _class_resolver(cls, prop, fallback, arg)._resolve_name return resolve_name, resolve_arg def _deferred_relationship(cls, prop): if isinstance(prop, RelationshipProperty): resolve_name, resolve_arg = _resolver(cls, prop) for attr in ( "order_by", "primaryjoin", "secondaryjoin", "secondary", "_user_defined_foreign_keys", "remote_side", ): v = getattr(prop, attr) if isinstance(v, util.string_types): setattr( prop, attr, resolve_arg(v, favor_tables=attr == "secondary"), ) for attr in ("argument",): v = getattr(prop, attr) if isinstance(v, util.string_types): setattr(prop, attr, resolve_name(v)) if prop.backref and isinstance(prop.backref, tuple): key, kwargs = prop.backref for attr in ( "primaryjoin", "secondaryjoin", "secondary", "foreign_keys", "remote_side", "order_by", ): if attr in kwargs and isinstance( kwargs[attr], util.string_types ): kwargs[attr] = resolve_arg(kwargs[attr]) return prop	12.27 KB	2026-01-20 12:58

Terminal