Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# ext/declarative/extensions.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
10"""Public API functions and helpers for declarative."""
12from __future__ import annotations
14import collections
15import contextlib
16from typing import Any
17from typing import Callable
18from typing import TYPE_CHECKING
19from typing import Union
21from ... import exc as sa_exc
22from ...engine import Connection
23from ...engine import Engine
24from ...orm import exc as orm_exc
25from ...orm import relationships
26from ...orm.base import _mapper_or_none
27from ...orm.clsregistry import _resolver
28from ...orm.decl_base import _DeferredMapperConfig
29from ...orm.util import polymorphic_union
30from ...schema import Table
31from ...util import OrderedDict
33if TYPE_CHECKING:
34 from ...sql.schema import MetaData
37class ConcreteBase:
38 """A helper class for 'concrete' declarative mappings.
40 :class:`.ConcreteBase` will use the :func:`.polymorphic_union`
41 function automatically, against all tables mapped as a subclass
42 to this class. The function is called via the
43 ``__declare_last__()`` function, which is essentially
44 a hook for the :meth:`.after_configured` event.
46 :class:`.ConcreteBase` produces a mapped
47 table for the class itself. Compare to :class:`.AbstractConcreteBase`,
48 which does not.
50 Example::
52 from sqlalchemy.ext.declarative import ConcreteBase
55 class Employee(ConcreteBase, Base):
56 __tablename__ = "employee"
57 employee_id = Column(Integer, primary_key=True)
58 name = Column(String(50))
59 __mapper_args__ = {
60 "polymorphic_identity": "employee",
61 "concrete": True,
62 }
65 class Manager(Employee):
66 __tablename__ = "manager"
67 employee_id = Column(Integer, primary_key=True)
68 name = Column(String(50))
69 manager_data = Column(String(40))
70 __mapper_args__ = {
71 "polymorphic_identity": "manager",
72 "concrete": True,
73 }
75 The name of the discriminator column used by :func:`.polymorphic_union`
76 defaults to the name ``type``. To suit the use case of a mapping where an
77 actual column in a mapped table is already named ``type``, the
78 discriminator name can be configured by setting the
79 ``_concrete_discriminator_name`` attribute::
81 class Employee(ConcreteBase, Base):
82 _concrete_discriminator_name = "_concrete_discriminator"
84 .. versionadded:: 1.3.19 Added the ``_concrete_discriminator_name``
85 attribute to :class:`_declarative.ConcreteBase` so that the
86 virtual discriminator column name can be customized.
88 .. versionchanged:: 1.4.2 The ``_concrete_discriminator_name`` attribute
89 need only be placed on the basemost class to take correct effect for
90 all subclasses. An explicit error message is now raised if the
91 mapped column names conflict with the discriminator name, whereas
92 in the 1.3.x series there would be some warnings and then a non-useful
93 query would be generated.
95 .. seealso::
97 :class:`.AbstractConcreteBase`
99 :ref:`concrete_inheritance`
102 """
104 @classmethod
105 def _create_polymorphic_union(cls, mappers, discriminator_name):
106 return polymorphic_union(
107 OrderedDict(
108 (mp.polymorphic_identity, mp.local_table) for mp in mappers
109 ),
110 discriminator_name,
111 "pjoin",
112 )
114 @classmethod
115 def __declare_first__(cls):
116 m = cls.__mapper__
117 if m.with_polymorphic:
118 return
120 discriminator_name = (
121 getattr(cls, "_concrete_discriminator_name", None) or "type"
122 )
124 mappers = list(m.self_and_descendants)
125 pjoin = cls._create_polymorphic_union(mappers, discriminator_name)
126 m._set_with_polymorphic(("*", pjoin))
127 m._set_polymorphic_on(pjoin.c[discriminator_name])
130class AbstractConcreteBase(ConcreteBase):
131 """A helper class for 'concrete' declarative mappings.
133 :class:`.AbstractConcreteBase` will use the :func:`.polymorphic_union`
134 function automatically, against all tables mapped as a subclass
135 to this class. The function is called via the
136 ``__declare_first__()`` function, which is essentially
137 a hook for the :meth:`.before_configured` event.
139 :class:`.AbstractConcreteBase` applies :class:`_orm.Mapper` for its
140 immediately inheriting class, as would occur for any other
141 declarative mapped class. However, the :class:`_orm.Mapper` is not
142 mapped to any particular :class:`.Table` object. Instead, it's
143 mapped directly to the "polymorphic" selectable produced by
144 :func:`.polymorphic_union`, and performs no persistence operations on its
145 own. Compare to :class:`.ConcreteBase`, which maps its
146 immediately inheriting class to an actual
147 :class:`.Table` that stores rows directly.
149 .. note::
151 The :class:`.AbstractConcreteBase` delays the mapper creation of the
152 base class until all the subclasses have been defined,
153 as it needs to create a mapping against a selectable that will include
154 all subclass tables. In order to achieve this, it waits for the
155 **mapper configuration event** to occur, at which point it scans
156 through all the configured subclasses and sets up a mapping that will
157 query against all subclasses at once.
159 While this event is normally invoked automatically, in the case of
160 :class:`.AbstractConcreteBase`, it may be necessary to invoke it
161 explicitly after **all** subclass mappings are defined, if the first
162 operation is to be a query against this base class. To do so, once all
163 the desired classes have been configured, the
164 :meth:`_orm.registry.configure` method on the :class:`_orm.registry`
165 in use can be invoked, which is available in relation to a particular
166 declarative base class::
168 Base.registry.configure()
170 Example::
172 from sqlalchemy.orm import DeclarativeBase
173 from sqlalchemy.ext.declarative import AbstractConcreteBase
176 class Base(DeclarativeBase):
177 pass
180 class Employee(AbstractConcreteBase, Base):
181 pass
184 class Manager(Employee):
185 __tablename__ = "manager"
186 employee_id = Column(Integer, primary_key=True)
187 name = Column(String(50))
188 manager_data = Column(String(40))
190 __mapper_args__ = {
191 "polymorphic_identity": "manager",
192 "concrete": True,
193 }
196 Base.registry.configure()
198 The abstract base class is handled by declarative in a special way;
199 at class configuration time, it behaves like a declarative mixin
200 or an ``__abstract__`` base class. Once classes are configured
201 and mappings are produced, it then gets mapped itself, but
202 after all of its descendants. This is a very unique system of mapping
203 not found in any other SQLAlchemy API feature.
205 Using this approach, we can specify columns and properties
206 that will take place on mapped subclasses, in the way that
207 we normally do as in :ref:`declarative_mixins`::
209 from sqlalchemy.ext.declarative import AbstractConcreteBase
212 class Company(Base):
213 __tablename__ = "company"
214 id = Column(Integer, primary_key=True)
217 class Employee(AbstractConcreteBase, Base):
218 strict_attrs = True
220 employee_id = Column(Integer, primary_key=True)
222 @declared_attr
223 def company_id(cls):
224 return Column(ForeignKey("company.id"))
226 @declared_attr
227 def company(cls):
228 return relationship("Company")
231 class Manager(Employee):
232 __tablename__ = "manager"
234 name = Column(String(50))
235 manager_data = Column(String(40))
237 __mapper_args__ = {
238 "polymorphic_identity": "manager",
239 "concrete": True,
240 }
243 Base.registry.configure()
245 When we make use of our mappings however, both ``Manager`` and
246 ``Employee`` will have an independently usable ``.company`` attribute::
248 session.execute(select(Employee).filter(Employee.company.has(id=5)))
250 :param strict_attrs: when specified on the base class, "strict" attribute
251 mode is enabled which attempts to limit ORM mapped attributes on the
252 base class to only those that are immediately present, while still
253 preserving "polymorphic" loading behavior.
255 .. versionadded:: 2.0
257 .. seealso::
259 :class:`.ConcreteBase`
261 :ref:`concrete_inheritance`
263 :ref:`abstract_concrete_base`
265 """
267 __no_table__ = True
269 @classmethod
270 def __declare_first__(cls):
271 cls._sa_decl_prepare_nocascade()
273 @classmethod
274 def _sa_decl_prepare_nocascade(cls):
275 if getattr(cls, "__mapper__", None):
276 return
278 to_map = _DeferredMapperConfig.config_for_cls(cls)
280 # can't rely on 'self_and_descendants' here
281 # since technically an immediate subclass
282 # might not be mapped, but a subclass
283 # may be.
284 mappers = []
285 stack = list(cls.__subclasses__())
286 while stack:
287 klass = stack.pop()
288 stack.extend(klass.__subclasses__())
289 mn = _mapper_or_none(klass)
290 if mn is not None:
291 mappers.append(mn)
293 discriminator_name = (
294 getattr(cls, "_concrete_discriminator_name", None) or "type"
295 )
296 pjoin = cls._create_polymorphic_union(mappers, discriminator_name)
298 # For columns that were declared on the class, these
299 # are normally ignored with the "__no_table__" mapping,
300 # unless they have a different attribute key vs. col name
301 # and are in the properties argument.
302 # In that case, ensure we update the properties entry
303 # to the correct column from the pjoin target table.
304 declared_cols = set(to_map.declared_columns)
305 declared_col_keys = {c.key for c in declared_cols}
306 for k, v in list(to_map.properties.items()):
307 if v in declared_cols:
308 to_map.properties[k] = pjoin.c[v.key]
309 declared_col_keys.remove(v.key)
311 to_map.local_table = pjoin
313 strict_attrs = cls.__dict__.get("strict_attrs", False)
315 m_args = to_map.mapper_args_fn or dict
317 def mapper_args():
318 args = m_args()
319 args["polymorphic_on"] = pjoin.c[discriminator_name]
320 args["polymorphic_abstract"] = True
321 if strict_attrs:
322 args["include_properties"] = (
323 set(pjoin.primary_key)
324 | declared_col_keys
325 | {discriminator_name}
326 )
327 args["with_polymorphic"] = ("*", pjoin)
328 return args
330 to_map.mapper_args_fn = mapper_args
332 to_map.map()
334 stack = [cls]
335 while stack:
336 scls = stack.pop(0)
337 stack.extend(scls.__subclasses__())
338 sm = _mapper_or_none(scls)
339 if sm and sm.concrete and sm.inherits is None:
340 for sup_ in scls.__mro__[1:]:
341 sup_sm = _mapper_or_none(sup_)
342 if sup_sm:
343 sm._set_concrete_base(sup_sm)
344 break
346 @classmethod
347 def _sa_raise_deferred_config(cls):
348 raise orm_exc.UnmappedClassError(
349 cls,
350 msg="Class %s is a subclass of AbstractConcreteBase and "
351 "has a mapping pending until all subclasses are defined. "
352 "Call the sqlalchemy.orm.configure_mappers() function after "
353 "all subclasses have been defined to "
354 "complete the mapping of this class."
355 % orm_exc._safe_cls_name(cls),
356 )
359class DeferredReflection:
360 """A helper class for construction of mappings based on
361 a deferred reflection step.
363 Normally, declarative can be used with reflection by
364 setting a :class:`_schema.Table` object using autoload_with=engine
365 as the ``__table__`` attribute on a declarative class.
366 The caveat is that the :class:`_schema.Table` must be fully
367 reflected, or at the very least have a primary key column,
368 at the point at which a normal declarative mapping is
369 constructed, meaning the :class:`_engine.Engine` must be available
370 at class declaration time.
372 The :class:`.DeferredReflection` mixin moves the construction
373 of mappers to be at a later point, after a specific
374 method is called which first reflects all :class:`_schema.Table`
375 objects created so far. Classes can define it as such::
377 from sqlalchemy.ext.declarative import declarative_base
378 from sqlalchemy.ext.declarative import DeferredReflection
380 Base = declarative_base()
383 class MyClass(DeferredReflection, Base):
384 __tablename__ = "mytable"
386 Above, ``MyClass`` is not yet mapped. After a series of
387 classes have been defined in the above fashion, all tables
388 can be reflected and mappings created using
389 :meth:`.prepare`::
391 engine = create_engine("someengine://...")
392 DeferredReflection.prepare(engine)
394 The :class:`.DeferredReflection` mixin can be applied to individual
395 classes, used as the base for the declarative base itself,
396 or used in a custom abstract class. Using an abstract base
397 allows that only a subset of classes to be prepared for a
398 particular prepare step, which is necessary for applications
399 that use more than one engine. For example, if an application
400 has two engines, you might use two bases, and prepare each
401 separately, e.g.::
403 class ReflectedOne(DeferredReflection, Base):
404 __abstract__ = True
407 class ReflectedTwo(DeferredReflection, Base):
408 __abstract__ = True
411 class MyClass(ReflectedOne):
412 __tablename__ = "mytable"
415 class MyOtherClass(ReflectedOne):
416 __tablename__ = "myothertable"
419 class YetAnotherClass(ReflectedTwo):
420 __tablename__ = "yetanothertable"
423 # ... etc.
425 Above, the class hierarchies for ``ReflectedOne`` and
426 ``ReflectedTwo`` can be configured separately::
428 ReflectedOne.prepare(engine_one)
429 ReflectedTwo.prepare(engine_two)
431 .. seealso::
433 :ref:`orm_declarative_reflected_deferred_reflection` - in the
434 :ref:`orm_declarative_table_config_toplevel` section.
436 """
438 @classmethod
439 def prepare(
440 cls, bind: Union[Engine, Connection], **reflect_kw: Any
441 ) -> None:
442 r"""Reflect all :class:`_schema.Table` objects for all current
443 :class:`.DeferredReflection` subclasses
445 :param bind: :class:`_engine.Engine` or :class:`_engine.Connection`
446 instance
448 ..versionchanged:: 2.0.16 a :class:`_engine.Connection` is also
449 accepted.
451 :param \**reflect_kw: additional keyword arguments passed to
452 :meth:`_schema.MetaData.reflect`, such as
453 :paramref:`_schema.MetaData.reflect.views`.
455 .. versionadded:: 2.0.16
457 """
459 to_map = _DeferredMapperConfig.classes_for_base(cls)
461 metadata_to_table = collections.defaultdict(set)
463 # first collect the primary __table__ for each class into a
464 # collection of metadata/schemaname -> table names
465 for thingy in to_map:
466 if thingy.local_table is not None:
467 metadata_to_table[
468 (thingy.local_table.metadata, thingy.local_table.schema)
469 ].add(thingy.local_table.name)
471 # then reflect all those tables into their metadatas
473 if isinstance(bind, Connection):
474 conn = bind
475 ctx = contextlib.nullcontext(enter_result=conn)
476 elif isinstance(bind, Engine):
477 ctx = bind.connect()
478 else:
479 raise sa_exc.ArgumentError(
480 f"Expected Engine or Connection, got {bind!r}"
481 )
483 with ctx as conn:
484 for (metadata, schema), table_names in metadata_to_table.items():
485 metadata.reflect(
486 conn,
487 only=table_names,
488 schema=schema,
489 extend_existing=True,
490 autoload_replace=False,
491 **reflect_kw,
492 )
494 metadata_to_table.clear()
496 # .map() each class, then go through relationships and look
497 # for secondary
498 for thingy in to_map:
499 thingy.map()
501 mapper = thingy.cls.__mapper__
502 metadata = mapper.class_.metadata
504 for rel in mapper._props.values():
505 if (
506 isinstance(rel, relationships.RelationshipProperty)
507 and rel._init_args.secondary._is_populated()
508 ):
509 secondary_arg = rel._init_args.secondary
511 if isinstance(secondary_arg.argument, Table):
512 secondary_table = secondary_arg.argument
513 metadata_to_table[
514 (
515 secondary_table.metadata,
516 secondary_table.schema,
517 )
518 ].add(secondary_table.name)
519 elif isinstance(secondary_arg.argument, str):
520 _, resolve_arg = _resolver(rel.parent.class_, rel)
522 resolver = resolve_arg(
523 secondary_arg.argument, True
524 )
525 metadata_to_table[
526 (metadata, thingy.local_table.schema)
527 ].add(secondary_arg.argument)
529 resolver._resolvers += (
530 cls._sa_deferred_table_resolver(metadata),
531 )
533 secondary_arg.argument = resolver()
535 for (metadata, schema), table_names in metadata_to_table.items():
536 metadata.reflect(
537 conn,
538 only=table_names,
539 schema=schema,
540 extend_existing=True,
541 autoload_replace=False,
542 )
544 @classmethod
545 def _sa_deferred_table_resolver(
546 cls, metadata: MetaData
547 ) -> Callable[[str], Table]:
548 def _resolve(key: str) -> Table:
549 # reflection has already occurred so this Table would have
550 # its contents already
551 return Table(key, metadata)
553 return _resolve
555 _sa_decl_prepare = True
557 @classmethod
558 def _sa_raise_deferred_config(cls):
559 raise orm_exc.UnmappedClassError(
560 cls,
561 msg="Class %s is a subclass of DeferredReflection. "
562 "Mappings are not produced until the .prepare() "
563 "method is called on the class hierarchy."
564 % orm_exc._safe_cls_name(cls),
565 )