cardinal_pythonlib.sqlalchemy.orm_inspect

Original code copyright (C) 2009-2022 Rudolf Cardinal (rudolf@pobox.com).

This file is part of cardinal_pythonlib.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Functions to inspect and copy SQLAlchemy ORM objects.

class cardinal_pythonlib.sqlalchemy.orm_inspect.SqlAlchemyAttrDictMixin[source]

Mixin to:

  • get a plain dictionary-like object (with attributes so we can use x.y rather than x['y']) from an SQLAlchemy ORM object

  • make a nice repr() default, maintaining field order

See https://stackoverflow.com/questions/2537471 and in particular https://stackoverflow.com/questions/2441796.

classmethod from_attrdict(attrdict: OrderedNamespace) → object[source]

Builds a new instance of the ORM object from values in an attrdict.

get_attrdict() → OrderedNamespace[source]

Returns what looks like a plain object with the values of the SQLAlchemy ORM object.

cardinal_pythonlib.sqlalchemy.orm_inspect.attrname_to_colname_dict(cls) → Dict[str, str][source]

Asks an SQLAlchemy class how its attribute names correspond to database column names.

Parameters:

cls – SQLAlchemy ORM class

Returns:

a dictionary mapping attribute names to database column names

cardinal_pythonlib.sqlalchemy.orm_inspect.coltype_as_typeengine(coltype: TraversibleType | TypeEngine) → TypeEngine[source]

Instances of SQLAlchemy column types are subclasses of TypeEngine. It’s possible to specify column types either as such instances, or as the class type. This function ensures that such classes are converted to instances.

To explain: you can specify columns like

a = Column("a", Integer)
b = Column("b", Integer())
c = Column("c", String(length=50))

isinstance(Integer, TypeEngine)  # False
isinstance(Integer(), TypeEngine)  # True
isinstance(String(length=50), TypeEngine)  # True

type(Integer)  # <class 'sqlalchemy.sql.visitors.VisitableType'>
type(Integer())  # <class 'sqlalchemy.sql.sqltypes.Integer'>
type(String)  # <class 'sqlalchemy.sql.visitors.VisitableType'>
type(String(length=50))  # <class 'sqlalchemy.sql.sqltypes.String'>

This function coerces things to a TypeEngine.

cardinal_pythonlib.sqlalchemy.orm_inspect.copy_sqla_object(obj: object, omit_fk: bool = True, omit_pk: bool = True, omit_attrs: List[str] | None = None, debug: bool = False) → object[source]

Given an SQLAlchemy object, creates a new object (FOR WHICH THE OBJECT MUST SUPPORT CREATION USING __init__() WITH NO PARAMETERS), and copies across all attributes, omitting PKs (by default), FKs (by default), and relationship attributes (always omitted).

Parameters:

  • obj – the object to copy

  • omit_fk – omit foreign keys (FKs)?

  • omit_pk – omit primary keys (PKs)?

  • omit_attrs – attributes (by name) not to copy

  • debug – be verbose

Returns:

a new copy of the object

cardinal_pythonlib.sqlalchemy.orm_inspect.deepcopy_sqla_object(startobj: object, session: Session, flush: bool = True, debug: bool = False, debug_walk: bool = False, debug_rewrite_rel: bool = False, objmap: Dict[object, object] | None = None) → object[source]

Makes a copy of the object, inserting it into session.

Uses deepcopy_sqla_objects() (q.v.).

A problem is the creation of duplicate dependency objects if you call it repeatedly.

Optionally, if you pass the objmap in (which maps old to new objects), you can call this function repeatedly to clone a related set of objects… … no, that doesn’t really work, as it doesn’t visit parents before children. The cardinal_pythonlib.sqlalchemy.merge_db.merge_db() function does that properly.

Parameters:
Returns:

the copied object matching startobj

cardinal_pythonlib.sqlalchemy.orm_inspect.deepcopy_sqla_objects(startobjs: List[object], session: Session, flush: bool = True, debug: bool = False, debug_walk: bool = True, debug_rewrite_rel: bool = False, objmap: Dict[object, object] | None = None) → None[source]

Makes a copy of the specified SQLAlchemy ORM objects, inserting them into a new session.

This function operates in several passes:

  1. Walk the ORM tree through all objects and their relationships, copying every object thus found (via copy_sqla_object(), without their relationships), and building a map from each source-session object to its equivalent destination-session object.

  2. Work through all the destination objects, rewriting their relationships (via rewrite_relationships()) so they relate to each other (rather than their source-session brethren).

  3. Insert all the destination-session objects into the destination session.

For this to succeed, every object must take an __init__ call with no arguments (see copy_sqla_object()). (We can’t specify the required args/kwargs, since we are copying a tree of arbitrary objects.)

Parameters:

  • startobjs – SQLAlchemy ORM objects to copy

  • session – destination SQLAlchemy Session into which to insert the copies

  • flush – flush the session when we’ve finished?

  • debug – be verbose?

  • debug_walk – be extra verbose when walking the ORM tree?

  • debug_rewrite_rel – be extra verbose when rewriting relationships?

  • objmap – starting object map from source-session to destination-session objects (see rewrite_relationships() for more detail); usually None to begin with.

cardinal_pythonlib.sqlalchemy.orm_inspect.gen_columns(obj) → Generator[Tuple[str, Column], None, None][source]

Asks a SQLAlchemy ORM object: “what are your SQLAlchemy columns?”

Yields tuples of (attr_name, Column) from an SQLAlchemy ORM object instance. Also works with the corresponding SQLAlchemy ORM class. Examples:

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.schema import Column
from sqlalchemy.sql.sqltypes import Integer

Base = declarative_base()

class MyClass(Base):
    __tablename__ = "mytable"
    pk = Column("pk", Integer, primary_key=True, autoincrement=True)
    a = Column("a", Integer)

x = MyClass()

list(gen_columns(x))
list(gen_columns(MyClass))
cardinal_pythonlib.sqlalchemy.orm_inspect.gen_columns_for_uninstrumented_class(cls: Type) → Generator[Tuple[str, Column], None, None][source]

Generate (attr_name, Column) tuples from an UNINSTRUMENTED class, i.e. one that does not inherit from declarative_base(). Use this for mixins of that kind.

SUBOPTIMAL. May produce warnings like:

SAWarning: Unmanaged access of declarative attribute id from non-mapped class GenericTabletRecordMixin

Try to use gen_columns() instead.

cardinal_pythonlib.sqlalchemy.orm_inspect.gen_orm_classes_from_base(base: Type) → Generator[Type, None, None][source]

From an SQLAlchemy ORM base class, yield all the subclasses (except those that are abstract).

If you begin with the proper :class`Base` class, then this should give all ORM classes in use.

cardinal_pythonlib.sqlalchemy.orm_inspect.gen_relationships(obj) → Generator[Tuple[str, RelationshipProperty, Type], None, None][source]

Yields tuples of (attrname, RelationshipProperty, related_class) for all relationships of an ORM object. The object ‘obj’ can be EITHER an instance OR a class.

cardinal_pythonlib.sqlalchemy.orm_inspect.get_metadata_from_orm_class_or_object(cls: Type) → MetaData[source]

Returns the MetaData object from an SQLAlchemy ORM class or instance.

cardinal_pythonlib.sqlalchemy.orm_inspect.get_orm_classes_by_table_name_from_base(base: Type) → Dict[str, Type][source]

Given an SQLAlchemy ORM base class, returns a dictionary whose keys are table names and whose values are ORM classes.

If you begin with the proper :class`Base` class, then this should give all tables and ORM classes in use.

cardinal_pythonlib.sqlalchemy.orm_inspect.get_orm_column_names(cls: Type, sort: bool = False) → List[str][source]

Gets column names (that is, database column names) from an SQLAlchemy ORM class.

cardinal_pythonlib.sqlalchemy.orm_inspect.get_orm_columns(cls: Type) → List[Column][source]

Gets Column objects from an SQLAlchemy ORM class. Does not provide their attribute names.

cardinal_pythonlib.sqlalchemy.orm_inspect.get_pk_attrnames(obj) → List[str][source]

Asks an SQLAlchemy ORM object: “what are your primary key(s)?”

Parameters:

obj – SQLAlchemy ORM object

Returns:

list of attribute names of primary-key columns

cardinal_pythonlib.sqlalchemy.orm_inspect.get_table_names_from_metadata(metadata: MetaData) → List[str][source]

Returns all database table names found in an SQLAlchemy MetaData object.

cardinal_pythonlib.sqlalchemy.orm_inspect.rewrite_relationships(oldobj: object, newobj: object, objmap: Dict[object, object], debug: bool = False, skip_table_names: List[str] | None = None) → None[source]

A utility function only. Used in copying objects between SQLAlchemy sessions.

Both oldobj and newobj are SQLAlchemy instances. The instance newobj is already a copy of oldobj but we wish to rewrite its relationships, according to the map objmap, which maps old to new objects.

For example:

  • Suppose a source session has a Customer record and a Sale record containing sale.customer_id, a foreign key to Customer.

  • We may have corresponding Python SQLAlchemy ORM objects customer_1_src and sale_1_src.

  • We copy them into a destination database, where their Python ORM objects are customer_1_dest and sale_1_dest.

  • In the process we set up an object map looking like:

    Old session     New session
-------------------------------
customer_1_src  customer_1_dest
sale_1_src      sale_1_dest

  • Now, we wish to make sale_1_dest have a relationship to customer_1_dest, in the same way that sale_1_src has a relationship to customer_1_src. This function will modify sale_1_dest accordingly, given this object map. It will observe that sale_1_src (here oldobj) has a relationship to customer_1_src; it will note that objmap maps customer_1_src to customer_1_dest; it will create the relationship from sale_1_dest (here newobj) to customer_1_dest.

Parameters:

  • oldobj – SQLAlchemy ORM object to read from

  • newobj – SQLAlchemy ORM object to write to

  • objmap – dictionary mapping “source” objects to their corresponding “destination” object.

  • debug – be verbose

  • skip_table_names – if a related table’s name is in this (optional) list, that relationship is skipped

cardinal_pythonlib.sqlalchemy.orm_inspect.walk_orm_tree(obj, debug: bool = False, seen: Set | None = None, skip_relationships_always: List[str] | None = None, skip_relationships_by_tablename: Dict[str, List[str]] | None = None, skip_all_relationships_for_tablenames: List[str] | None = None, skip_all_objects_for_tablenames: List[str] | None = None) → Generator[object, None, None][source]

Starting with a SQLAlchemy ORM object, this function walks a relationship tree, yielding each of the objects once.

To skip attributes by name, put the attribute name(s) in skip_attrs_always. To skip by table name, pass skip_attrs_by_tablename as e.g.

{'sometable': ['attr1_to_skip', 'attr2_to_skip']}
Parameters:

  • obj – the SQLAlchemy ORM object to walk

  • debug – be verbose

  • seen – usually None, but can be a set of objects marked as “already seen”; if an object is in this set, it is skipped

  • skip_relationships_always – relationships are skipped if the relationship has a name in this (optional) list

  • skip_relationships_by_tablename – optional dictionary mapping table names (keys) to relationship attribute names (values); if the “related table”/”relationship attribute” pair are in this dictionary, the relationship is skipped

  • skip_all_relationships_for_tablenames – relationships are skipped if the the related table has a name in this (optional) list

  • skip_all_objects_for_tablenames – if the object belongs to a table whose name is in this (optional) list, the object is skipped

Yields:

SQLAlchemy ORM objects (including the starting object)