Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/cardinal_pythonlib/sqlalchemy/orm_inspect.py : 25%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#!/usr/bin/env python
2# cardinal_pythonlib/sqlalchemy/orm_inspect.py
4"""
5===============================================================================
7 Original code copyright (C) 2009-2021 Rudolf Cardinal (rudolf@pobox.com).
9 This file is part of cardinal_pythonlib.
11 Licensed under the Apache License, Version 2.0 (the "License");
12 you may not use this file except in compliance with the License.
13 You may obtain a copy of the License at
15 https://www.apache.org/licenses/LICENSE-2.0
17 Unless required by applicable law or agreed to in writing, software
18 distributed under the License is distributed on an "AS IS" BASIS,
19 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 See the License for the specific language governing permissions and
21 limitations under the License.
23===============================================================================
25**Functions to inspect and copy SQLAlchemy ORM objects.**
27"""
29from typing import (Dict, Generator, List, Set, Tuple, Type, TYPE_CHECKING,
30 Union)
32# noinspection PyProtectedMember
33from sqlalchemy.ext.declarative.base import _get_immediate_cls_attr
34from sqlalchemy.inspection import inspect
35from sqlalchemy.orm.base import class_mapper
36from sqlalchemy.orm.mapper import Mapper
37from sqlalchemy.orm.relationships import RelationshipProperty
38from sqlalchemy.orm.session import Session
39from sqlalchemy.sql.schema import Column, MetaData
40from sqlalchemy.sql.type_api import TypeEngine
41from sqlalchemy.sql.visitors import VisitableType
42from sqlalchemy.util import OrderedProperties
44from cardinal_pythonlib.classes import gen_all_subclasses
45from cardinal_pythonlib.enumlike import OrderedNamespace
46from cardinal_pythonlib.dicts import reversedict
47from cardinal_pythonlib.logs import get_brace_style_log_with_null_handler
49if TYPE_CHECKING:
50 from sqlalchemy.orm.state import InstanceState
51 from sqlalchemy.sql.schema import Table
53log = get_brace_style_log_with_null_handler(__name__)
56# =============================================================================
57# Creating ORM objects conveniently, etc.
58# =============================================================================
60def coltype_as_typeengine(coltype: Union[VisitableType,
61 TypeEngine]) -> TypeEngine:
62 """
63 Instances of SQLAlchemy column types are subclasses of ``TypeEngine``.
64 It's possible to specify column types either as such instances, or as the
65 class type. This function ensures that such classes are converted to
66 instances.
68 To explain: you can specify columns like
70 .. code-block:: python
72 a = Column("a", Integer)
73 b = Column("b", Integer())
74 c = Column("c", String(length=50))
76 isinstance(Integer, TypeEngine) # False
77 isinstance(Integer(), TypeEngine) # True
78 isinstance(String(length=50), TypeEngine) # True
80 type(Integer) # <class 'sqlalchemy.sql.visitors.VisitableType'>
81 type(Integer()) # <class 'sqlalchemy.sql.sqltypes.Integer'>
82 type(String) # <class 'sqlalchemy.sql.visitors.VisitableType'>
83 type(String(length=50)) # <class 'sqlalchemy.sql.sqltypes.String'>
85 This function coerces things to a :class:`TypeEngine`.
86 """
87 if isinstance(coltype, TypeEngine):
88 return coltype
89 return coltype() # type: TypeEngine
92# =============================================================================
93# SqlAlchemyAttrDictMixin
94# =============================================================================
96class SqlAlchemyAttrDictMixin(object):
97 """
98 Mixin to:
100 - get a plain dictionary-like object (with attributes so we can use ``x.y``
101 rather than ``x['y']``) from an SQLAlchemy ORM object
102 - make a nice ``repr()`` default, maintaining field order
104 See https://stackoverflow.com/questions/2537471 and in particular
105 https://stackoverflow.com/questions/2441796.
106 """
108 def get_attrdict(self) -> OrderedNamespace:
109 """
110 Returns what looks like a plain object with the values of the
111 SQLAlchemy ORM object.
112 """
113 # noinspection PyUnresolvedReferences
114 columns = self.__table__.columns.keys()
115 values = (getattr(self, x) for x in columns)
116 zipped = zip(columns, values)
117 return OrderedNamespace(zipped)
119 def __repr__(self) -> str:
120 return "<{classname}({kvp})>".format(
121 classname=type(self).__name__,
122 kvp=", ".join(f"{k}={v!r}"
123 for k, v in self.get_attrdict().items())
124 )
126 @classmethod
127 def from_attrdict(cls, attrdict: OrderedNamespace) -> object:
128 """
129 Builds a new instance of the ORM object from values in an attrdict.
130 """
131 dictionary = attrdict.__dict__
132 # noinspection PyArgumentList
133 return cls(**dictionary)
136# =============================================================================
137# Traverse ORM relationships (SQLAlchemy ORM)
138# =============================================================================
140def walk_orm_tree(obj,
141 debug: bool = False,
142 seen: Set = None,
143 skip_relationships_always: List[str] = None,
144 skip_relationships_by_tablename: Dict[str, List[str]] = None,
145 skip_all_relationships_for_tablenames: List[str] = None,
146 skip_all_objects_for_tablenames: List[str] = None) \
147 -> Generator[object, None, None]:
148 """
149 Starting with a SQLAlchemy ORM object, this function walks a
150 relationship tree, yielding each of the objects once.
152 To skip attributes by name, put the attribute name(s) in
153 ``skip_attrs_always``. To skip by table name, pass
154 ``skip_attrs_by_tablename`` as e.g.
156 .. code-block:: python
158 {'sometable': ['attr1_to_skip', 'attr2_to_skip']}
161 Args:
162 obj: the SQLAlchemy ORM object to walk
164 debug: be verbose
166 seen: usually ``None``, but can be a set of objects marked as "already
167 seen"; if an object is in this set, it is skipped
169 skip_relationships_always: relationships are skipped if the
170 relationship has a name in this (optional) list
172 skip_relationships_by_tablename: optional dictionary mapping table
173 names (keys) to relationship attribute names (values); if the
174 "related table"/"relationship attribute" pair are in this
175 dictionary, the relationship is skipped
177 skip_all_relationships_for_tablenames: relationships are skipped if the
178 the related table has a name in this (optional) list
180 skip_all_objects_for_tablenames: if the object belongs to a table whose
181 name is in this (optional) list, the object is skipped
183 Yields:
184 SQLAlchemy ORM objects (including the starting object)
186 """
187 # http://docs.sqlalchemy.org/en/latest/faq/sessions.html#faq-walk-objects
188 skip_relationships_always = skip_relationships_always or [] # type: List[str] # noqa
189 skip_relationships_by_tablename = skip_relationships_by_tablename or {} # type: Dict[str, List[str]] # noqa
190 skip_all_relationships_for_tablenames = skip_all_relationships_for_tablenames or [] # type: List[str] # noqa
191 skip_all_objects_for_tablenames = skip_all_objects_for_tablenames or [] # type: List[str] # noqa
192 stack = [obj]
193 if seen is None:
194 seen = set()
195 while stack:
196 obj = stack.pop(0)
197 if obj in seen:
198 continue
199 tablename = obj.__tablename__
200 if tablename in skip_all_objects_for_tablenames:
201 continue
202 seen.add(obj)
203 if debug:
204 log.debug("walk: yielding {!r}", obj)
205 yield obj
206 insp = inspect(obj) # type: InstanceState
207 for relationship in insp.mapper.relationships: # type: RelationshipProperty # noqa
208 attrname = relationship.key
209 # Skip?
210 if attrname in skip_relationships_always:
211 continue
212 if tablename in skip_all_relationships_for_tablenames:
213 continue
214 if (tablename in skip_relationships_by_tablename and
215 attrname in skip_relationships_by_tablename[tablename]):
216 continue
217 # Process relationship
218 if debug:
219 log.debug("walk: following relationship {}", relationship)
220 related = getattr(obj, attrname)
221 if debug and related:
222 log.debug("walk: queueing {!r}", related)
223 if relationship.uselist:
224 stack.extend(related)
225 elif related is not None:
226 stack.append(related)
229# =============================================================================
230# deepcopy an SQLAlchemy object
231# =============================================================================
232# Use case: object X is in the database; we want to clone it to object Y,
233# which we can then save to the database, i.e. copying all SQLAlchemy field
234# attributes of X except its PK. We also want it to copy anything that is
235# dependent upon X, i.e. traverse relationships.
236#
237# https://groups.google.com/forum/#!topic/sqlalchemy/wb2M_oYkQdY
238# https://groups.google.com/forum/#!searchin/sqlalchemy/cascade%7Csort:date/sqlalchemy/eIOkkXwJ-Ms/JLnpI2wJAAAJ # noqa
240def copy_sqla_object(obj: object,
241 omit_fk: bool = True,
242 omit_pk: bool = True,
243 omit_attrs: List[str] = None,
244 debug: bool = False) -> object:
245 """
246 Given an SQLAlchemy object, creates a new object (FOR WHICH THE OBJECT
247 MUST SUPPORT CREATION USING ``__init__()`` WITH NO PARAMETERS), and copies
248 across all attributes, omitting PKs (by default), FKs (by default), and
249 relationship attributes (always omitted).
251 Args:
252 obj: the object to copy
253 omit_fk: omit foreign keys (FKs)?
254 omit_pk: omit primary keys (PKs)?
255 omit_attrs: attributes (by name) not to copy
256 debug: be verbose
258 Returns:
259 a new copy of the object
260 """
261 omit_attrs = omit_attrs or [] # type: List[str]
262 cls = type(obj)
263 mapper = class_mapper(cls)
264 newobj = cls() # not: cls.__new__(cls)
265 rel_keys = set([c.key for c in mapper.relationships])
266 prohibited = rel_keys
267 if omit_pk:
268 pk_keys = set([c.key for c in mapper.primary_key])
269 prohibited |= pk_keys
270 if omit_fk:
271 fk_keys = set([c.key for c in mapper.columns if c.foreign_keys])
272 prohibited |= fk_keys
273 prohibited |= set(omit_attrs)
274 if debug:
275 log.debug("copy_sqla_object: skipping: {}", prohibited)
276 for k in [p.key for p in mapper.iterate_properties
277 if p.key not in prohibited]:
278 try:
279 value = getattr(obj, k)
280 if debug:
281 log.debug("copy_sqla_object: processing attribute {} = {}",
282 k, value)
283 setattr(newobj, k, value)
284 except AttributeError:
285 if debug:
286 log.debug("copy_sqla_object: failed attribute {}", k)
287 pass
288 return newobj
291def rewrite_relationships(oldobj: object,
292 newobj: object,
293 objmap: Dict[object, object],
294 debug: bool = False,
295 skip_table_names: List[str] = None) -> None:
296 """
297 A utility function only.
298 Used in copying objects between SQLAlchemy sessions.
300 Both ``oldobj`` and ``newobj`` are SQLAlchemy instances. The instance
301 ``newobj`` is already a copy of ``oldobj`` but we wish to rewrite its
302 relationships, according to the map ``objmap``, which maps old to new
303 objects.
305 For example:
307 - Suppose a source session has a Customer record and a Sale record
308 containing ``sale.customer_id``, a foreign key to Customer.
310 - We may have corresponding Python SQLAlchemy ORM objects
311 ``customer_1_src`` and ``sale_1_src``.
313 - We copy them into a destination database, where their Python ORM objects
314 are ``customer_1_dest`` and ``sale_1_dest``.
316 - In the process we set up an object map looking like:
318 .. code-block:: none
320 Old session New session
321 -------------------------------
322 customer_1_src customer_1_dest
323 sale_1_src sale_1_dest
325 - Now, we wish to make ``sale_1_dest`` have a relationship to
326 ``customer_1_dest``, in the same way that ``sale_1_src`` has a
327 relationship to ``customer_1_src``. This function will modify
328 ``sale_1_dest`` accordingly, given this object map. It will observe that
329 ``sale_1_src`` (here ``oldobj``) has a relationship to
330 ``customer_1_src``; it will note that ``objmap`` maps ``customer_1_src``
331 to ``customer_1_dest``; it will create the relationship from
332 ``sale_1_dest`` (here ``newobj``) to ``customer_1_dest``.
334 Args:
335 oldobj: SQLAlchemy ORM object to read from
337 newobj: SQLAlchemy ORM object to write to
339 objmap: dictionary mapping "source" objects to their corresponding
340 "destination" object.
342 debug: be verbose
344 skip_table_names: if a related table's name is in this (optional) list,
345 that relationship is skipped
346 """
347 skip_table_names = skip_table_names or [] # type: List[str]
348 insp = inspect(oldobj) # type: InstanceState
349 # insp.mapper.relationships is of type
350 # sqlalchemy.utils._collections.ImmutableProperties, which is basically
351 # a sort of AttrDict.
352 for attrname_rel in insp.mapper.relationships.items(): # type: Tuple[str, RelationshipProperty] # noqa
353 attrname = attrname_rel[0]
354 rel_prop = attrname_rel[1]
355 if rel_prop.viewonly:
356 if debug:
357 log.debug("Skipping viewonly relationship")
358 continue # don't attempt to write viewonly relationships # noqa
359 related_class = rel_prop.mapper.class_
360 related_table_name = related_class.__tablename__ # type: str
361 if related_table_name in skip_table_names:
362 if debug:
363 log.debug("Skipping relationship for related table {!r}",
364 related_table_name)
365 continue
366 # The relationship is an abstract object (so getting the
367 # relationship from the old object and from the new, with e.g.
368 # newrel = newinsp.mapper.relationships[oldrel.key],
369 # yield the same object. All we need from it is the key name.
370 # rel_key = rel.key # type: str
371 # ... but also available from the mapper as attrname, above
372 related_old = getattr(oldobj, attrname)
373 if rel_prop.uselist:
374 related_new = [objmap[r] for r in related_old]
375 elif related_old is not None:
376 related_new = objmap[related_old]
377 else:
378 related_new = None
379 if debug:
380 log.debug("rewrite_relationships: relationship {} -> {}",
381 attrname, related_new)
382 setattr(newobj, attrname, related_new)
385def deepcopy_sqla_objects(
386 startobjs: List[object],
387 session: Session,
388 flush: bool = True,
389 debug: bool = False,
390 debug_walk: bool = True,
391 debug_rewrite_rel: bool = False,
392 objmap: Dict[object, object] = None) -> None:
393 """
394 Makes a copy of the specified SQLAlchemy ORM objects, inserting them into a
395 new session.
397 This function operates in several passes:
399 1. Walk the ORM tree through all objects and their relationships, copying
400 every object thus found (via :func:`copy_sqla_object`, without their
401 relationships), and building a map from each source-session object to
402 its equivalent destination-session object.
404 2. Work through all the destination objects, rewriting their relationships
405 (via :func:`rewrite_relationships`) so they relate to each other (rather
406 than their source-session brethren).
408 3. Insert all the destination-session objects into the destination session.
410 For this to succeed, every object must take an ``__init__`` call with no
411 arguments (see :func:`copy_sqla_object`). (We can't specify the required
412 ``args``/``kwargs``, since we are copying a tree of arbitrary objects.)
414 Args:
415 startobjs: SQLAlchemy ORM objects to copy
416 session: destination SQLAlchemy :class:`Session` into which to insert
417 the copies
418 flush: flush the session when we've finished?
419 debug: be verbose?
420 debug_walk: be extra verbose when walking the ORM tree?
421 debug_rewrite_rel: be extra verbose when rewriting relationships?
422 objmap: starting object map from source-session to destination-session
423 objects (see :func:`rewrite_relationships` for more detail);
424 usually ``None`` to begin with.
425 """
426 if objmap is None:
427 objmap = {} # keys = old objects, values = new objects
428 if debug:
429 log.debug("deepcopy_sqla_objects: pass 1: create new objects")
431 # Pass 1: iterate through all objects. (Can't guarantee to get
432 # relationships correct until we've done this, since we don't know whether
433 # or where the "root" of the PK tree is.)
434 seen = set()
435 for startobj in startobjs:
436 for oldobj in walk_orm_tree(startobj, seen=seen, debug=debug_walk):
437 if debug:
438 log.debug("deepcopy_sqla_objects: copying {}", oldobj)
439 newobj = copy_sqla_object(oldobj, omit_pk=True, omit_fk=True)
440 # Don't insert the new object into the session here; it may trigger
441 # an autoflush as the relationships are queried, and the new
442 # objects are not ready for insertion yet (as their relationships
443 # aren't set).
444 # Note also the session.no_autoflush option:
445 # "sqlalchemy.exc.OperationalError: (raised as a result of Query-
446 # invoked autoflush; consider using a session.no_autoflush block if
447 # this flush is occurring prematurely)..."
448 objmap[oldobj] = newobj
450 # Pass 2: set all relationship properties.
451 if debug:
452 log.debug("deepcopy_sqla_objects: pass 2: set relationships")
453 for oldobj, newobj in objmap.items():
454 if debug:
455 log.debug("deepcopy_sqla_objects: newobj: {}", newobj)
456 rewrite_relationships(oldobj, newobj, objmap, debug=debug_rewrite_rel)
458 # Now we can do session insert.
459 if debug:
460 log.debug("deepcopy_sqla_objects: pass 3: insert into session")
461 for newobj in objmap.values():
462 session.add(newobj)
464 # Done
465 if debug:
466 log.debug("deepcopy_sqla_objects: done")
467 if flush:
468 session.flush()
471def deepcopy_sqla_object(startobj: object,
472 session: Session,
473 flush: bool = True,
474 debug: bool = False,
475 debug_walk: bool = False,
476 debug_rewrite_rel: bool = False,
477 objmap: Dict[object, object] = None) -> object:
478 """
479 Makes a copy of the object, inserting it into ``session``.
481 Uses :func:`deepcopy_sqla_objects` (q.v.).
483 A problem is the creation of duplicate dependency objects if you call it
484 repeatedly.
486 Optionally, if you pass the objmap in (which maps old to new objects), you
487 can call this function repeatedly to clone a related set of objects...
488 ... no, that doesn't really work, as it doesn't visit parents before
489 children. The :func:`cardinal_pythonlib.sqlalchemy.merge_db.merge_db`
490 function does that properly.
492 Args:
493 startobj: SQLAlchemy ORM object to deep-copy
494 session: see :func:`deepcopy_sqla_objects`
495 flush: see :func:`deepcopy_sqla_objects`
496 debug: see :func:`deepcopy_sqla_objects`
497 debug_walk: see :func:`deepcopy_sqla_objects`
498 debug_rewrite_rel: see :func:`deepcopy_sqla_objects`
499 objmap: see :func:`deepcopy_sqla_objects`
501 Returns:
502 the copied object matching ``startobj``
504 """
505 if objmap is None:
506 objmap = {} # keys = old objects, values = new objects
507 deepcopy_sqla_objects(
508 startobjs=[startobj],
509 session=session,
510 flush=flush,
511 debug=debug,
512 debug_walk=debug_walk,
513 debug_rewrite_rel=debug_rewrite_rel,
514 objmap=objmap
515 )
516 return objmap[startobj] # returns the new object matching startobj
519# =============================================================================
520# Get Columns from an ORM instance
521# =============================================================================
523def gen_columns(obj) -> Generator[Tuple[str, Column], None, None]:
524 """
525 Asks a SQLAlchemy ORM object: "what are your SQLAlchemy columns?"
527 Yields tuples of ``(attr_name, Column)`` from an SQLAlchemy ORM object
528 instance. Also works with the corresponding SQLAlchemy ORM class. Examples:
530 .. code-block:: python
532 from sqlalchemy.ext.declarative import declarative_base
533 from sqlalchemy.sql.schema import Column
534 from sqlalchemy.sql.sqltypes import Integer
536 Base = declarative_base()
538 class MyClass(Base):
539 __tablename__ = "mytable"
540 pk = Column("pk", Integer, primary_key=True, autoincrement=True)
541 a = Column("a", Integer)
543 x = MyClass()
545 list(gen_columns(x))
546 list(gen_columns(MyClass))
548 """
549 mapper = obj.__mapper__ # type: Mapper
550 assert mapper, (
551 f"gen_columns called on {obj!r} which is not an SQLAlchemy ORM object"
552 )
553 colmap = mapper.columns # type: OrderedProperties
554 if not colmap:
555 return
556 for attrname, column in colmap.items():
557 # NB: column.name is the SQL column name, not the attribute name
558 yield attrname, column
559 # Don't bother using
560 # cls = obj.__class_
561 # for attrname in dir(cls):
562 # cls_attr = getattr(cls, attrname)
563 # # ... because, for columns, these will all be instances of
564 # # sqlalchemy.orm.attributes.InstrumentedAttribute.
567def get_pk_attrnames(obj) -> List[str]:
568 """
569 Asks an SQLAlchemy ORM object: "what are your primary key(s)?"
571 Args:
572 obj: SQLAlchemy ORM object
574 Returns:
575 list of attribute names of primary-key columns
577 """
578 return [attrname
579 for attrname, column in gen_columns(obj)
580 if column.primary_key]
583def gen_columns_for_uninstrumented_class(cls: Type) \
584 -> Generator[Tuple[str, Column], None, None]:
585 """
586 Generate ``(attr_name, Column)`` tuples from an UNINSTRUMENTED class, i.e.
587 one that does not inherit from ``declarative_base()``. Use this for mixins
588 of that kind.
590 SUBOPTIMAL. May produce warnings like:
592 .. code-block:: none
594 SAWarning: Unmanaged access of declarative attribute id from non-mapped class GenericTabletRecordMixin
596 Try to use :func:`gen_columns` instead.
597 """ # noqa
598 for attrname in dir(cls):
599 potential_column = getattr(cls, attrname)
600 if isinstance(potential_column, Column):
601 yield attrname, potential_column
604def attrname_to_colname_dict(cls) -> Dict[str, str]:
605 """
606 Asks an SQLAlchemy class how its attribute names correspond to database
607 column names.
609 Args:
610 cls: SQLAlchemy ORM class
612 Returns:
613 a dictionary mapping attribute names to database column names
614 """
615 attr_col = {} # type: Dict[str, str]
616 for attrname, column in gen_columns(cls):
617 attr_col[attrname] = column.name
618 return attr_col
621def colname_to_attrname_dict(cls) -> Dict[str, str]:
622 return reversedict(attrname_to_colname_dict(cls))
625# =============================================================================
626# Get relationships from an ORM instance
627# =============================================================================
629def gen_relationships(obj) -> Generator[Tuple[str, RelationshipProperty, Type],
630 None, None]:
631 """
632 Yields tuples of ``(attrname, RelationshipProperty, related_class)``
633 for all relationships of an ORM object.
634 The object 'obj' can be EITHER an instance OR a class.
635 """
636 insp = inspect(obj) # type: InstanceState
637 # insp.mapper.relationships is of type
638 # sqlalchemy.utils._collections.ImmutableProperties, which is basically
639 # a sort of AttrDict.
640 for attrname, rel_prop in insp.mapper.relationships.items(): # type: Tuple[str, RelationshipProperty] # noqa
641 # noinspection PyUnresolvedReferences
642 related_class = rel_prop.mapper.class_
643 # log.critical("gen_relationships: attrname={!r}, "
644 # "rel_prop={!r}, related_class={!r}, rel_prop.info={!r}",
645 # attrname, rel_prop, related_class, rel_prop.info)
646 yield attrname, rel_prop, related_class
649# =============================================================================
650# Inspect ORM objects (SQLAlchemy ORM)
651# =============================================================================
653def get_orm_columns(cls: Type) -> List[Column]:
654 """
655 Gets :class:`Column` objects from an SQLAlchemy ORM class.
656 Does not provide their attribute names.
657 """
658 mapper = inspect(cls) # type: Mapper
659 # ... returns InstanceState if called with an ORM object
660 # http://docs.sqlalchemy.org/en/latest/orm/session_state_management.html#session-object-states # noqa
661 # ... returns Mapper if called with an ORM class
662 # http://docs.sqlalchemy.org/en/latest/orm/mapping_api.html#sqlalchemy.orm.mapper.Mapper # noqa
663 colmap = mapper.columns # type: OrderedProperties
664 return colmap.values()
667def get_orm_column_names(cls: Type, sort: bool = False) -> List[str]:
668 """
669 Gets column names (that is, database column names) from an SQLAlchemy
670 ORM class.
671 """
672 colnames = [col.name for col in get_orm_columns(cls)]
673 return sorted(colnames) if sort else colnames
676# =============================================================================
677# Inspect metadata (SQLAlchemy ORM)
678# =============================================================================
680def get_table_names_from_metadata(metadata: MetaData) -> List[str]:
681 """
682 Returns all database table names found in an SQLAlchemy :class:`MetaData`
683 object.
684 """
685 return [table.name for table in metadata.tables.values()]
688def get_metadata_from_orm_class_or_object(cls: Type) -> MetaData:
689 """
690 Returns the :class:`MetaData` object from an SQLAlchemy ORM class or
691 instance.
692 """
693 # noinspection PyUnresolvedReferences
694 table = cls.__table__ # type: Table
695 return table.metadata
698def gen_orm_classes_from_base(base: Type) -> Generator[Type, None, None]:
699 """
700 From an SQLAlchemy ORM base class, yield all the subclasses (except those
701 that are abstract).
703 If you begin with the proper :class`Base` class, then this should give all
704 ORM classes in use.
705 """
706 for cls in gen_all_subclasses(base):
707 if _get_immediate_cls_attr(cls, '__abstract__', strict=True):
708 # This is SQLAlchemy's own way of detecting abstract classes; see
709 # sqlalchemy.ext.declarative.base
710 continue # NOT an ORM class
711 yield cls
714def get_orm_classes_by_table_name_from_base(base: Type) -> Dict[str, Type]:
715 """
716 Given an SQLAlchemy ORM base class, returns a dictionary whose keys are
717 table names and whose values are ORM classes.
719 If you begin with the proper :class`Base` class, then this should give all
720 tables and ORM classes in use.
721 """
722 # noinspection PyUnresolvedReferences
723 return {cls.__tablename__: cls for cls in gen_orm_classes_from_base(base)}