codekingpro/portable-devtools
115k
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
8
9
10"""Public API functions and helpers for declarative."""
11from __future__ import annotations
12
13import collections
14import contextlib
15from typing import Any
16from typing import Callable
17from typing import TYPE_CHECKING
18from typing import Union
19
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
31
32if TYPE_CHECKING:
33 from ...sql.schema import MetaData
34
35
36class ConcreteBase:
37 """A helper class for 'concrete' declarative mappings.
38
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.
44
45 :class:`.ConcreteBase` produces a mapped
46 table for the class itself. Compare to :class:`.AbstractConcreteBase`,
47 which does not.
48
49 Example::
50
51 from sqlalchemy.ext.declarative import ConcreteBase
52
53
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 }
62
63
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 }
73
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::
79
80 class Employee(ConcreteBase, Base):
81 _concrete_discriminator_name = "_concrete_discriminator"
82
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.
86
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.
93
94 .. seealso::
95
96 :class:`.AbstractConcreteBase`
97
98 :ref:`concrete_inheritance`
99
100
101 """
102
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 )
112
113 @classmethod
114 def __declare_first__(cls):
115 m = cls.__mapper__
116 if m.with_polymorphic:
117 return
118
119 discriminator_name = (
120 getattr(cls, "_concrete_discriminator_name", None) or "type"
121 )
122
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])
127
128
129class AbstractConcreteBase(ConcreteBase):
130 """A helper class for 'concrete' declarative mappings.
131
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.
137
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.
147
148 .. note::
149
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.
157
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::
166
167 Base.registry.configure()
168
169 Example::
170
171 from sqlalchemy.orm import DeclarativeBase
172 from sqlalchemy.ext.declarative import AbstractConcreteBase
173
174
175 class Base(DeclarativeBase):
176 pass
177
178
179 class Employee(AbstractConcreteBase, Base):
180 pass
181
182
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))
188
189 __mapper_args__ = {
190 "polymorphic_identity": "manager",
191 "concrete": True,
192 }
193
194
195 Base.registry.configure()
196
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.
203
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`::
207
208 from sqlalchemy.ext.declarative import AbstractConcreteBase
209
210
211 class Company(Base):
212 __tablename__ = "company"
213 id = Column(Integer, primary_key=True)
214
215
216 class Employee(AbstractConcreteBase, Base):
217 strict_attrs = True
218
219 employee_id = Column(Integer, primary_key=True)
220
221 @declared_attr
222 def company_id(cls):
223 return Column(ForeignKey("company.id"))
224
225 @declared_attr
226 def company(cls):
227 return relationship("Company")
228
229
230 class Manager(Employee):
231 __tablename__ = "manager"
232
233 name = Column(String(50))
234 manager_data = Column(String(40))
235
236 __mapper_args__ = {
237 "polymorphic_identity": "manager",
238 "concrete": True,
239 }
240
241
242 Base.registry.configure()
243
244 When we make use of our mappings however, both ``Manager`` and
245 ``Employee`` will have an independently usable ``.company`` attribute::
246
247 session.execute(select(Employee).filter(Employee.company.has(id=5)))
248
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.
253
254 .. versionadded:: 2.0
255
256 .. seealso::
257
258 :class:`.ConcreteBase`
259
260 :ref:`concrete_inheritance`
261
262 :ref:`abstract_concrete_base`
263
264 """
265
266 __no_table__ = True
267
268 @classmethod
269 def __declare_first__(cls):
270 cls._sa_decl_prepare_nocascade()
271
272 @classmethod
273 def _sa_decl_prepare_nocascade(cls):
274 if getattr(cls, "__mapper__", None):
275 return
276
277 to_map = _DeferredMapperConfig.config_for_cls(cls)
278
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)
291
292 discriminator_name = (
293 getattr(cls, "_concrete_discriminator_name", None) or "type"
294 )
295 pjoin = cls._create_polymorphic_union(mappers, discriminator_name)
296
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)
309
310 to_map.local_table = pjoin
311
312 strict_attrs = cls.__dict__.get("strict_attrs", False)
313
314 m_args = to_map.mapper_args_fn or dict
315
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
328
329 to_map.mapper_args_fn = mapper_args
330
331 to_map.map()
332
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
344
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 )
356
357
358class DeferredReflection:
359 """A helper class for construction of mappings based on
360 a deferred reflection step.
361
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.
370
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::
375
376 from sqlalchemy.ext.declarative import declarative_base
377 from sqlalchemy.ext.declarative import DeferredReflection
378
379 Base = declarative_base()
380
381
382 class MyClass(DeferredReflection, Base):
383 __tablename__ = "mytable"
384
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`::
389
390 engine = create_engine("someengine://...")
391 DeferredReflection.prepare(engine)
392
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.::
401
402 class ReflectedOne(DeferredReflection, Base):
403 __abstract__ = True
404
405
406 class ReflectedTwo(DeferredReflection, Base):
407 __abstract__ = True
408
409
410 class MyClass(ReflectedOne):
411 __tablename__ = "mytable"
412
413
414 class MyOtherClass(ReflectedOne):
415 __tablename__ = "myothertable"
416
417
418 class YetAnotherClass(ReflectedTwo):
419 __tablename__ = "yetanothertable"
420
421
422 # ... etc.
423
424 Above, the class hierarchies for ``ReflectedOne`` and
425 ``ReflectedTwo`` can be configured separately::
426
427 ReflectedOne.prepare(engine_one)
428 ReflectedTwo.prepare(engine_two)
429
430 .. seealso::
431
432 :ref:`orm_declarative_reflected_deferred_reflection` - in the
433 :ref:`orm_declarative_table_config_toplevel` section.
434
435 """
436
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
443
444 :param bind: :class:`_engine.Engine` or :class:`_engine.Connection`
445 instance
446
447 ..versionchanged:: 2.0.16 a :class:`_engine.Connection` is also
448 accepted.
449
450 :param \**reflect_kw: additional keyword arguments passed to
451 :meth:`_schema.MetaData.reflect`, such as
452 :paramref:`_schema.MetaData.reflect.views`.
453
454 .. versionadded:: 2.0.16
455
456 """
457
458 to_map = _DeferredMapperConfig.classes_for_base(cls)
459
460 metadata_to_table = collections.defaultdict(set)
461
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)
469
470 # then reflect all those tables into their metadatas
471
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 )
481
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 )
492
493 metadata_to_table.clear()
494
495 # .map() each class, then go through relationships and look
496 # for secondary
497 for thingy in to_map:
498 thingy.map()
499
500 mapper = thingy.cls.__mapper__
501 metadata = mapper.class_.metadata
502
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
509
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)
520
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)
527
528 resolver._resolvers += (
529 cls._sa_deferred_table_resolver(metadata),
530 )
531
532 secondary_arg.argument = resolver()
533
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 )
542
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)
551
552 return _resolve
553
554 _sa_decl_prepare = True
555
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 )
565