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 _DeferredDeclarativeConfig
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:`.MapperEvents.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 .. versionchanged:: 1.4.2 The ``_concrete_discriminator_name`` attribute
85 need only be placed on the basemost class to take correct effect for
86 all subclasses. An explicit error message is now raised if the
87 mapped column names conflict with the discriminator name, whereas
88 in the 1.3.x series there would be some warnings and then a non-useful
89 query would be generated.
91 .. seealso::
93 :class:`.AbstractConcreteBase`
95 :ref:`concrete_inheritance`
98 """
100 @classmethod
101 def _create_polymorphic_union(cls, mappers, discriminator_name):
102 return polymorphic_union(
103 OrderedDict(
104 (mp.polymorphic_identity, mp.local_table) for mp in mappers
105 ),
106 discriminator_name,
107 "pjoin",
108 )
110 @classmethod
111 def __declare_first__(cls):
112 m = cls.__mapper__
113 if m.with_polymorphic:
114 return
116 discriminator_name = (
117 getattr(cls, "_concrete_discriminator_name", None) or "type"
118 )
120 mappers = list(m.self_and_descendants)
121 pjoin = cls._create_polymorphic_union(mappers, discriminator_name)
122 m._set_with_polymorphic(("*", pjoin))
123 m._set_polymorphic_on(pjoin.c[discriminator_name])
126class AbstractConcreteBase(ConcreteBase):
127 """A helper class for 'concrete' declarative mappings.
129 :class:`.AbstractConcreteBase` will use the :func:`.polymorphic_union`
130 function automatically, against all tables mapped as a subclass
131 to this class. The function is called via the
132 ``__declare_first__()`` function, which is essentially
133 a hook for the :meth:`.MapperEvents.before_configured` event.
135 :class:`.AbstractConcreteBase` applies :class:`_orm.Mapper` for its
136 immediately inheriting class, as would occur for any other
137 declarative mapped class. However, the :class:`_orm.Mapper` is not
138 mapped to any particular :class:`.Table` object. Instead, it's
139 mapped directly to the "polymorphic" selectable produced by
140 :func:`.polymorphic_union`, and performs no persistence operations on its
141 own. Compare to :class:`.ConcreteBase`, which maps its
142 immediately inheriting class to an actual
143 :class:`.Table` that stores rows directly.
145 .. note::
147 The :class:`.AbstractConcreteBase` delays the mapper creation of the
148 base class until all the subclasses have been defined,
149 as it needs to create a mapping against a selectable that will include
150 all subclass tables. In order to achieve this, it waits for the
151 **mapper configuration event** to occur, at which point it scans
152 through all the configured subclasses and sets up a mapping that will
153 query against all subclasses at once.
155 While this event is normally invoked automatically, in the case of
156 :class:`.AbstractConcreteBase`, it may be necessary to invoke it
157 explicitly after **all** subclass mappings are defined, if the first
158 operation is to be a query against this base class. To do so, once all
159 the desired classes have been configured, the
160 :meth:`_orm.registry.configure` method on the :class:`_orm.registry`
161 in use can be invoked, which is available in relation to a particular
162 declarative base class::
164 Base.registry.configure()
166 Example::
168 from sqlalchemy.orm import DeclarativeBase
169 from sqlalchemy.ext.declarative import AbstractConcreteBase
172 class Base(DeclarativeBase):
173 pass
176 class Employee(AbstractConcreteBase, Base):
177 pass
180 class Manager(Employee):
181 __tablename__ = "manager"
182 employee_id = Column(Integer, primary_key=True)
183 name = Column(String(50))
184 manager_data = Column(String(40))
186 __mapper_args__ = {
187 "polymorphic_identity": "manager",
188 "concrete": True,
189 }
192 Base.registry.configure()
194 The abstract base class is handled by declarative in a special way;
195 at class configuration time, it behaves like a declarative mixin
196 or an ``__abstract__`` base class. Once classes are configured
197 and mappings are produced, it then gets mapped itself, but
198 after all of its descendants. This is a very unique system of mapping
199 not found in any other SQLAlchemy API feature.
201 Using this approach, we can specify columns and properties
202 that will take place on mapped subclasses, in the way that
203 we normally do as in :ref:`declarative_mixins`::
205 from sqlalchemy.ext.declarative import AbstractConcreteBase
208 class Company(Base):
209 __tablename__ = "company"
210 id = Column(Integer, primary_key=True)
213 class Employee(AbstractConcreteBase, Base):
214 strict_attrs = True
216 employee_id = Column(Integer, primary_key=True)
218 @declared_attr
219 def company_id(cls):
220 return Column(ForeignKey("company.id"))
222 @declared_attr
223 def company(cls):
224 return relationship("Company")
227 class Manager(Employee):
228 __tablename__ = "manager"
230 name = Column(String(50))
231 manager_data = Column(String(40))
233 __mapper_args__ = {
234 "polymorphic_identity": "manager",
235 "concrete": True,
236 }
239 Base.registry.configure()
241 When we make use of our mappings however, both ``Manager`` and
242 ``Employee`` will have an independently usable ``.company`` attribute::
244 session.execute(select(Employee).filter(Employee.company.has(id=5)))
246 :param strict_attrs: when specified on the base class, "strict" attribute
247 mode is enabled which attempts to limit ORM mapped attributes on the
248 base class to only those that are immediately present, while still
249 preserving "polymorphic" loading behavior.
251 .. versionadded:: 2.0
253 .. seealso::
255 :class:`.ConcreteBase`
257 :ref:`concrete_inheritance`
259 :ref:`abstract_concrete_base`
261 """
263 __no_table__ = True
265 @classmethod
266 def __declare_first__(cls):
267 cls._sa_decl_prepare_nocascade()
269 @classmethod
270 def _sa_decl_prepare_nocascade(cls):
271 if getattr(cls, "__mapper__", None):
272 return
274 to_map = _DeferredDeclarativeConfig.config_for_cls(cls)
276 # can't rely on 'self_and_descendants' here
277 # since technically an immediate subclass
278 # might not be mapped, but a subclass
279 # may be.
280 mappers = []
281 stack = list(cls.__subclasses__())
282 while stack:
283 klass = stack.pop()
284 stack.extend(klass.__subclasses__())
285 mn = _mapper_or_none(klass)
286 if mn is not None:
287 mappers.append(mn)
289 discriminator_name = (
290 getattr(cls, "_concrete_discriminator_name", None) or "type"
291 )
292 pjoin = cls._create_polymorphic_union(mappers, discriminator_name)
294 # For columns that were declared on the class, these
295 # are normally ignored with the "__no_table__" mapping,
296 # unless they have a different attribute key vs. col name
297 # and are in the properties argument.
298 # In that case, ensure we update the properties entry
299 # to the correct column from the pjoin target table.
300 declared_cols = set(to_map.declared_columns)
301 declared_col_keys = {c.key for c in declared_cols}
302 for k, v in list(to_map.properties.items()):
303 if v in declared_cols:
304 to_map.properties[k] = pjoin.c[v.key]
305 declared_col_keys.remove(v.key)
307 to_map.local_table = pjoin
309 strict_attrs = cls.__dict__.get("strict_attrs", False)
311 m_args = to_map.mapper_args_fn or dict
313 def mapper_args():
314 args = m_args()
315 args["polymorphic_on"] = pjoin.c[discriminator_name]
316 args["polymorphic_abstract"] = True
317 if strict_attrs:
318 args["include_properties"] = (
319 set(pjoin.primary_key)
320 | declared_col_keys
321 | {discriminator_name}
322 )
323 args["with_polymorphic"] = ("*", pjoin)
324 return args
326 to_map.mapper_args_fn = mapper_args
328 to_map.map()
330 stack = [cls]
331 while stack:
332 scls = stack.pop(0)
333 stack.extend(scls.__subclasses__())
334 sm = _mapper_or_none(scls)
335 if sm and sm.concrete and sm.inherits is None:
336 for sup_ in scls.__mro__[1:]:
337 sup_sm = _mapper_or_none(sup_)
338 if sup_sm:
339 sm._set_concrete_base(sup_sm)
340 break
342 @classmethod
343 def _sa_raise_deferred_config(cls):
344 raise orm_exc.UnmappedClassError(
345 cls,
346 msg="Class %s is a subclass of AbstractConcreteBase and "
347 "has a mapping pending until all subclasses are defined. "
348 "Call the sqlalchemy.orm.configure_mappers() function after "
349 "all subclasses have been defined to "
350 "complete the mapping of this class."
351 % orm_exc._safe_cls_name(cls),
352 )
355class DeferredReflection:
356 """A helper class for construction of mappings based on
357 a deferred reflection step.
359 Normally, declarative can be used with reflection by
360 setting a :class:`_schema.Table` object using autoload_with=engine
361 as the ``__table__`` attribute on a declarative class.
362 The caveat is that the :class:`_schema.Table` must be fully
363 reflected, or at the very least have a primary key column,
364 at the point at which a normal declarative mapping is
365 constructed, meaning the :class:`_engine.Engine` must be available
366 at class declaration time.
368 The :class:`.DeferredReflection` mixin moves the construction
369 of mappers to be at a later point, after a specific
370 method is called which first reflects all :class:`_schema.Table`
371 objects created so far. Classes can define it as such::
373 from sqlalchemy.ext.declarative import declarative_base
374 from sqlalchemy.ext.declarative import DeferredReflection
376 Base = declarative_base()
379 class MyClass(DeferredReflection, Base):
380 __tablename__ = "mytable"
382 Above, ``MyClass`` is not yet mapped. After a series of
383 classes have been defined in the above fashion, all tables
384 can be reflected and mappings created using
385 :meth:`.prepare`::
387 engine = create_engine("someengine://...")
388 DeferredReflection.prepare(engine)
390 The :class:`.DeferredReflection` mixin can be applied to individual
391 classes, used as the base for the declarative base itself,
392 or used in a custom abstract class. Using an abstract base
393 allows that only a subset of classes to be prepared for a
394 particular prepare step, which is necessary for applications
395 that use more than one engine. For example, if an application
396 has two engines, you might use two bases, and prepare each
397 separately, e.g.::
399 class ReflectedOne(DeferredReflection, Base):
400 __abstract__ = True
403 class ReflectedTwo(DeferredReflection, Base):
404 __abstract__ = True
407 class MyClass(ReflectedOne):
408 __tablename__ = "mytable"
411 class MyOtherClass(ReflectedOne):
412 __tablename__ = "myothertable"
415 class YetAnotherClass(ReflectedTwo):
416 __tablename__ = "yetanothertable"
419 # ... etc.
421 Above, the class hierarchies for ``ReflectedOne`` and
422 ``ReflectedTwo`` can be configured separately::
424 ReflectedOne.prepare(engine_one)
425 ReflectedTwo.prepare(engine_two)
427 .. seealso::
429 :ref:`orm_declarative_reflected_deferred_reflection` - in the
430 :ref:`orm_declarative_table_config_toplevel` section.
432 """
434 @classmethod
435 def prepare(
436 cls, bind: Union[Engine, Connection], **reflect_kw: Any
437 ) -> None:
438 r"""Reflect all :class:`_schema.Table` objects for all current
439 :class:`.DeferredReflection` subclasses
441 :param bind: :class:`_engine.Engine` or :class:`_engine.Connection`
442 instance
444 ..versionchanged:: 2.0.16 a :class:`_engine.Connection` is also
445 accepted.
447 :param \**reflect_kw: additional keyword arguments passed to
448 :meth:`_schema.MetaData.reflect`, such as
449 :paramref:`_schema.MetaData.reflect.views`.
451 .. versionadded:: 2.0.16
453 """
455 to_map = _DeferredDeclarativeConfig.classes_for_base(cls)
457 metadata_to_table = collections.defaultdict(set)
459 # first collect the primary __table__ for each class into a
460 # collection of metadata/schemaname -> table names
461 for thingy in to_map:
462 if thingy.local_table is not None:
463 metadata_to_table[
464 (thingy.local_table.metadata, thingy.local_table.schema)
465 ].add(thingy.local_table.name)
467 # then reflect all those tables into their metadatas
469 if isinstance(bind, Connection):
470 conn = bind
471 ctx = contextlib.nullcontext(enter_result=conn)
472 elif isinstance(bind, Engine):
473 ctx = bind.connect()
474 else:
475 raise sa_exc.ArgumentError(
476 f"Expected Engine or Connection, got {bind!r}"
477 )
479 with ctx as conn:
480 for (metadata, schema), table_names in metadata_to_table.items():
481 metadata.reflect(
482 conn,
483 only=table_names,
484 schema=schema,
485 extend_existing=True,
486 autoload_replace=False,
487 **reflect_kw,
488 )
490 metadata_to_table.clear()
492 # .map() each class, then go through relationships and look
493 # for secondary
494 for thingy in to_map:
495 thingy.map()
497 mapper = thingy.cls.__mapper__
498 metadata = mapper.class_.metadata
500 for rel in mapper._props.values():
501 if (
502 isinstance(rel, relationships.RelationshipProperty)
503 and rel._init_args.secondary._is_populated()
504 ):
505 secondary_arg = rel._init_args.secondary
507 if isinstance(secondary_arg.argument, Table):
508 secondary_table = secondary_arg.argument
509 metadata_to_table[
510 (
511 secondary_table.metadata,
512 secondary_table.schema,
513 )
514 ].add(secondary_table.name)
515 elif isinstance(secondary_arg.argument, str):
516 _, resolve_arg = _resolver(rel.parent.class_, rel)
518 resolver = resolve_arg(
519 secondary_arg.argument, True
520 )
521 metadata_to_table[
522 (metadata, thingy.local_table.schema)
523 ].add(secondary_arg.argument)
525 resolver._resolvers += (
526 cls._sa_deferred_table_resolver(metadata),
527 )
529 secondary_arg.argument = resolver()
531 for (metadata, schema), table_names in metadata_to_table.items():
532 metadata.reflect(
533 conn,
534 only=table_names,
535 schema=schema,
536 extend_existing=True,
537 autoload_replace=False,
538 )
540 @classmethod
541 def _sa_deferred_table_resolver(
542 cls, metadata: MetaData
543 ) -> Callable[[str], Table]:
544 def _resolve(key: str) -> Table:
545 # reflection has already occurred so this Table would have
546 # its contents already
547 return Table(key, metadata)
549 return _resolve
551 _sa_decl_prepare = True
553 @classmethod
554 def _sa_raise_deferred_config(cls):
555 raise orm_exc.UnmappedClassError(
556 cls,
557 msg="Class %s is a subclass of DeferredReflection. "
558 "Mappings are not produced until the .prepare() "
559 "method is called on the class hierarchy."
560 % orm_exc._safe_cls_name(cls),
561 )