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-2025 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."""
11from __future__ import annotations
13import collections
14import contextlib
15from typing import Any
16from typing import Callable
17from typing import TYPE_CHECKING
18from typing import Union
20from ... import exc as sa_exc
21from ...engine import Connection
22from ...engine import Engine
23from ...orm import exc as orm_exc
24from ...orm import relationships
25from ...orm.base import _mapper_or_none
26from ...orm.clsregistry import _resolver
27from ...orm.decl_base import _DeferredMapperConfig
28from ...orm.util import polymorphic_union
29from ...schema import Table
30from ...util import OrderedDict
32if TYPE_CHECKING:
33 from ...sql.schema import MetaData
36class ConcreteBase:
37 """A helper class for 'concrete' declarative mappings.
39 :class:`.ConcreteBase` will use the :func:`.polymorphic_union`
40 function automatically, against all tables mapped as a subclass
41 to this class. The function is called via the
42 ``__declare_last__()`` function, which is essentially
43 a hook for the :meth:`.after_configured` event.
45 :class:`.ConcreteBase` produces a mapped
46 table for the class itself. Compare to :class:`.AbstractConcreteBase`,
47 which does not.
49 Example::
51 from sqlalchemy.ext.declarative import ConcreteBase
54 class Employee(ConcreteBase, Base):
55 __tablename__ = "employee"
56 employee_id = Column(Integer, primary_key=True)
57 name = Column(String(50))
58 __mapper_args__ = {
59 "polymorphic_identity": "employee",
60 "concrete": True,
61 }
64 class Manager(Employee):
65 __tablename__ = "manager"
66 employee_id = Column(Integer, primary_key=True)
67 name = Column(String(50))
68 manager_data = Column(String(40))
69 __mapper_args__ = {
70 "polymorphic_identity": "manager",
71 "concrete": True,
72 }
74 The name of the discriminator column used by :func:`.polymorphic_union`
75 defaults to the name ``type``. To suit the use case of a mapping where an
76 actual column in a mapped table is already named ``type``, the
77 discriminator name can be configured by setting the
78 ``_concrete_discriminator_name`` attribute::
80 class Employee(ConcreteBase, Base):
81 _concrete_discriminator_name = "_concrete_discriminator"
83 .. versionadded:: 1.3.19 Added the ``_concrete_discriminator_name``
84 attribute to :class:`_declarative.ConcreteBase` so that the
85 virtual discriminator column name can be customized.
87 .. versionchanged:: 1.4.2 The ``_concrete_discriminator_name`` attribute
88 need only be placed on the basemost class to take correct effect for
89 all subclasses. An explicit error message is now raised if the
90 mapped column names conflict with the discriminator name, whereas
91 in the 1.3.x series there would be some warnings and then a non-useful
92 query would be generated.
94 .. seealso::
96 :class:`.AbstractConcreteBase`
98 :ref:`concrete_inheritance`
101 """
103 @classmethod
104 def _create_polymorphic_union(cls, mappers, discriminator_name):
105 return polymorphic_union(
106 OrderedDict(
107 (mp.polymorphic_identity, mp.local_table) for mp in mappers
108 ),
109 discriminator_name,
110 "pjoin",
111 )
113 @classmethod
114 def __declare_first__(cls):
115 m = cls.__mapper__
116 if m.with_polymorphic:
117 return
119 discriminator_name = (
120 getattr(cls, "_concrete_discriminator_name", None) or "type"
121 )
123 mappers = list(m.self_and_descendants)
124 pjoin = cls._create_polymorphic_union(mappers, discriminator_name)
125 m._set_with_polymorphic(("*", pjoin))
126 m._set_polymorphic_on(pjoin.c[discriminator_name])
129class AbstractConcreteBase(ConcreteBase):
130 """A helper class for 'concrete' declarative mappings.
132 :class:`.AbstractConcreteBase` will use the :func:`.polymorphic_union`
133 function automatically, against all tables mapped as a subclass
134 to this class. The function is called via the
135 ``__declare_first__()`` function, which is essentially
136 a hook for the :meth:`.before_configured` event.
138 :class:`.AbstractConcreteBase` applies :class:`_orm.Mapper` for its
139 immediately inheriting class, as would occur for any other
140 declarative mapped class. However, the :class:`_orm.Mapper` is not
141 mapped to any particular :class:`.Table` object. Instead, it's
142 mapped directly to the "polymorphic" selectable produced by
143 :func:`.polymorphic_union`, and performs no persistence operations on its
144 own. Compare to :class:`.ConcreteBase`, which maps its
145 immediately inheriting class to an actual
146 :class:`.Table` that stores rows directly.
148 .. note::
150 The :class:`.AbstractConcreteBase` delays the mapper creation of the
151 base class until all the subclasses have been defined,
152 as it needs to create a mapping against a selectable that will include
153 all subclass tables. In order to achieve this, it waits for the
154 **mapper configuration event** to occur, at which point it scans
155 through all the configured subclasses and sets up a mapping that will
156 query against all subclasses at once.
158 While this event is normally invoked automatically, in the case of
159 :class:`.AbstractConcreteBase`, it may be necessary to invoke it
160 explicitly after **all** subclass mappings are defined, if the first
161 operation is to be a query against this base class. To do so, once all
162 the desired classes have been configured, the
163 :meth:`_orm.registry.configure` method on the :class:`_orm.registry`
164 in use can be invoked, which is available in relation to a particular
165 declarative base class::
167 Base.registry.configure()
169 Example::
171 from sqlalchemy.orm import DeclarativeBase
172 from sqlalchemy.ext.declarative import AbstractConcreteBase
175 class Base(DeclarativeBase):
176 pass
179 class Employee(AbstractConcreteBase, Base):
180 pass
183 class Manager(Employee):
184 __tablename__ = "manager"
185 employee_id = Column(Integer, primary_key=True)
186 name = Column(String(50))
187 manager_data = Column(String(40))
189 __mapper_args__ = {
190 "polymorphic_identity": "manager",
191 "concrete": True,
192 }
195 Base.registry.configure()
197 The abstract base class is handled by declarative in a special way;
198 at class configuration time, it behaves like a declarative mixin
199 or an ``__abstract__`` base class. Once classes are configured
200 and mappings are produced, it then gets mapped itself, but
201 after all of its descendants. This is a very unique system of mapping
202 not found in any other SQLAlchemy API feature.
204 Using this approach, we can specify columns and properties
205 that will take place on mapped subclasses, in the way that
206 we normally do as in :ref:`declarative_mixins`::
208 from sqlalchemy.ext.declarative import AbstractConcreteBase
211 class Company(Base):
212 __tablename__ = "company"
213 id = Column(Integer, primary_key=True)
216 class Employee(AbstractConcreteBase, Base):
217 strict_attrs = True
219 employee_id = Column(Integer, primary_key=True)
221 @declared_attr
222 def company_id(cls):
223 return Column(ForeignKey("company.id"))
225 @declared_attr
226 def company(cls):
227 return relationship("Company")
230 class Manager(Employee):
231 __tablename__ = "manager"
233 name = Column(String(50))
234 manager_data = Column(String(40))
236 __mapper_args__ = {
237 "polymorphic_identity": "manager",
238 "concrete": True,
239 }
242 Base.registry.configure()
244 When we make use of our mappings however, both ``Manager`` and
245 ``Employee`` will have an independently usable ``.company`` attribute::
247 session.execute(select(Employee).filter(Employee.company.has(id=5)))
249 :param strict_attrs: when specified on the base class, "strict" attribute
250 mode is enabled which attempts to limit ORM mapped attributes on the
251 base class to only those that are immediately present, while still
252 preserving "polymorphic" loading behavior.
254 .. versionadded:: 2.0
256 .. seealso::
258 :class:`.ConcreteBase`
260 :ref:`concrete_inheritance`
262 :ref:`abstract_concrete_base`
264 """
266 __no_table__ = True
268 @classmethod
269 def __declare_first__(cls):
270 cls._sa_decl_prepare_nocascade()
272 @classmethod
273 def _sa_decl_prepare_nocascade(cls):
274 if getattr(cls, "__mapper__", None):
275 return
277 to_map = _DeferredMapperConfig.config_for_cls(cls)
279 # can't rely on 'self_and_descendants' here
280 # since technically an immediate subclass
281 # might not be mapped, but a subclass
282 # may be.
283 mappers = []
284 stack = list(cls.__subclasses__())
285 while stack:
286 klass = stack.pop()
287 stack.extend(klass.__subclasses__())
288 mn = _mapper_or_none(klass)
289 if mn is not None:
290 mappers.append(mn)
292 discriminator_name = (
293 getattr(cls, "_concrete_discriminator_name", None) or "type"
294 )
295 pjoin = cls._create_polymorphic_union(mappers, discriminator_name)
297 # For columns that were declared on the class, these
298 # are normally ignored with the "__no_table__" mapping,
299 # unless they have a different attribute key vs. col name
300 # and are in the properties argument.
301 # In that case, ensure we update the properties entry
302 # to the correct column from the pjoin target table.
303 declared_cols = set(to_map.declared_columns)
304 declared_col_keys = {c.key for c in declared_cols}
305 for k, v in list(to_map.properties.items()):
306 if v in declared_cols:
307 to_map.properties[k] = pjoin.c[v.key]
308 declared_col_keys.remove(v.key)
310 to_map.local_table = pjoin
312 strict_attrs = cls.__dict__.get("strict_attrs", False)
314 m_args = to_map.mapper_args_fn or dict
316 def mapper_args():
317 args = m_args()
318 args["polymorphic_on"] = pjoin.c[discriminator_name]
319 args["polymorphic_abstract"] = True
320 if strict_attrs:
321 args["include_properties"] = (
322 set(pjoin.primary_key)
323 | declared_col_keys
324 | {discriminator_name}
325 )
326 args["with_polymorphic"] = ("*", pjoin)
327 return args
329 to_map.mapper_args_fn = mapper_args
331 to_map.map()
333 stack = [cls]
334 while stack:
335 scls = stack.pop(0)
336 stack.extend(scls.__subclasses__())
337 sm = _mapper_or_none(scls)
338 if sm and sm.concrete and sm.inherits is None:
339 for sup_ in scls.__mro__[1:]:
340 sup_sm = _mapper_or_none(sup_)
341 if sup_sm:
342 sm._set_concrete_base(sup_sm)
343 break
345 @classmethod
346 def _sa_raise_deferred_config(cls):
347 raise orm_exc.UnmappedClassError(
348 cls,
349 msg="Class %s is a subclass of AbstractConcreteBase and "
350 "has a mapping pending until all subclasses are defined. "
351 "Call the sqlalchemy.orm.configure_mappers() function after "
352 "all subclasses have been defined to "
353 "complete the mapping of this class."
354 % orm_exc._safe_cls_name(cls),
355 )
358class DeferredReflection:
359 """A helper class for construction of mappings based on
360 a deferred reflection step.
362 Normally, declarative can be used with reflection by
363 setting a :class:`_schema.Table` object using autoload_with=engine
364 as the ``__table__`` attribute on a declarative class.
365 The caveat is that the :class:`_schema.Table` must be fully
366 reflected, or at the very least have a primary key column,
367 at the point at which a normal declarative mapping is
368 constructed, meaning the :class:`_engine.Engine` must be available
369 at class declaration time.
371 The :class:`.DeferredReflection` mixin moves the construction
372 of mappers to be at a later point, after a specific
373 method is called which first reflects all :class:`_schema.Table`
374 objects created so far. Classes can define it as such::
376 from sqlalchemy.ext.declarative import declarative_base
377 from sqlalchemy.ext.declarative import DeferredReflection
379 Base = declarative_base()
382 class MyClass(DeferredReflection, Base):
383 __tablename__ = "mytable"
385 Above, ``MyClass`` is not yet mapped. After a series of
386 classes have been defined in the above fashion, all tables
387 can be reflected and mappings created using
388 :meth:`.prepare`::
390 engine = create_engine("someengine://...")
391 DeferredReflection.prepare(engine)
393 The :class:`.DeferredReflection` mixin can be applied to individual
394 classes, used as the base for the declarative base itself,
395 or used in a custom abstract class. Using an abstract base
396 allows that only a subset of classes to be prepared for a
397 particular prepare step, which is necessary for applications
398 that use more than one engine. For example, if an application
399 has two engines, you might use two bases, and prepare each
400 separately, e.g.::
402 class ReflectedOne(DeferredReflection, Base):
403 __abstract__ = True
406 class ReflectedTwo(DeferredReflection, Base):
407 __abstract__ = True
410 class MyClass(ReflectedOne):
411 __tablename__ = "mytable"
414 class MyOtherClass(ReflectedOne):
415 __tablename__ = "myothertable"
418 class YetAnotherClass(ReflectedTwo):
419 __tablename__ = "yetanothertable"
422 # ... etc.
424 Above, the class hierarchies for ``ReflectedOne`` and
425 ``ReflectedTwo`` can be configured separately::
427 ReflectedOne.prepare(engine_one)
428 ReflectedTwo.prepare(engine_two)
430 .. seealso::
432 :ref:`orm_declarative_reflected_deferred_reflection` - in the
433 :ref:`orm_declarative_table_config_toplevel` section.
435 """
437 @classmethod
438 def prepare(
439 cls, bind: Union[Engine, Connection], **reflect_kw: Any
440 ) -> None:
441 r"""Reflect all :class:`_schema.Table` objects for all current
442 :class:`.DeferredReflection` subclasses
444 :param bind: :class:`_engine.Engine` or :class:`_engine.Connection`
445 instance
447 ..versionchanged:: 2.0.16 a :class:`_engine.Connection` is also
448 accepted.
450 :param \**reflect_kw: additional keyword arguments passed to
451 :meth:`_schema.MetaData.reflect`, such as
452 :paramref:`_schema.MetaData.reflect.views`.
454 .. versionadded:: 2.0.16
456 """
458 to_map = _DeferredMapperConfig.classes_for_base(cls)
460 metadata_to_table = collections.defaultdict(set)
462 # first collect the primary __table__ for each class into a
463 # collection of metadata/schemaname -> table names
464 for thingy in to_map:
465 if thingy.local_table is not None:
466 metadata_to_table[
467 (thingy.local_table.metadata, thingy.local_table.schema)
468 ].add(thingy.local_table.name)
470 # then reflect all those tables into their metadatas
472 if isinstance(bind, Connection):
473 conn = bind
474 ctx = contextlib.nullcontext(enter_result=conn)
475 elif isinstance(bind, Engine):
476 ctx = bind.connect()
477 else:
478 raise sa_exc.ArgumentError(
479 f"Expected Engine or Connection, got {bind!r}"
480 )
482 with ctx as conn:
483 for (metadata, schema), table_names in metadata_to_table.items():
484 metadata.reflect(
485 conn,
486 only=table_names,
487 schema=schema,
488 extend_existing=True,
489 autoload_replace=False,
490 **reflect_kw,
491 )
493 metadata_to_table.clear()
495 # .map() each class, then go through relationships and look
496 # for secondary
497 for thingy in to_map:
498 thingy.map()
500 mapper = thingy.cls.__mapper__
501 metadata = mapper.class_.metadata
503 for rel in mapper._props.values():
504 if (
505 isinstance(rel, relationships.RelationshipProperty)
506 and rel._init_args.secondary._is_populated()
507 ):
508 secondary_arg = rel._init_args.secondary
510 if isinstance(secondary_arg.argument, Table):
511 secondary_table = secondary_arg.argument
512 metadata_to_table[
513 (
514 secondary_table.metadata,
515 secondary_table.schema,
516 )
517 ].add(secondary_table.name)
518 elif isinstance(secondary_arg.argument, str):
519 _, resolve_arg = _resolver(rel.parent.class_, rel)
521 resolver = resolve_arg(
522 secondary_arg.argument, True
523 )
524 metadata_to_table[
525 (metadata, thingy.local_table.schema)
526 ].add(secondary_arg.argument)
528 resolver._resolvers += (
529 cls._sa_deferred_table_resolver(metadata),
530 )
532 secondary_arg.argument = resolver()
534 for (metadata, schema), table_names in metadata_to_table.items():
535 metadata.reflect(
536 conn,
537 only=table_names,
538 schema=schema,
539 extend_existing=True,
540 autoload_replace=False,
541 )
543 @classmethod
544 def _sa_deferred_table_resolver(
545 cls, metadata: MetaData
546 ) -> Callable[[str], Table]:
547 def _resolve(key: str) -> Table:
548 # reflection has already occurred so this Table would have
549 # its contents already
550 return Table(key, metadata)
552 return _resolve
554 _sa_decl_prepare = True
556 @classmethod
557 def _sa_raise_deferred_config(cls):
558 raise orm_exc.UnmappedClassError(
559 cls,
560 msg="Class %s is a subclass of DeferredReflection. "
561 "Mappings are not produced until the .prepare() "
562 "method is called on the class hierarchy."
563 % orm_exc._safe_cls_name(cls),
564 )