SQLAlchemy reflect ORM classes declared on the same Base or share ORM classes between instances

Question

I have a problem with trying to share ORM classes between different Instances of a database accessing class. The gist of it is, that in the setup phase of my program a database engine and a DeclarativeBase Object are created which will later be injected whenever I need to create a new object that requires a database connection. The different instances will need to share the ORM class definitions some of the time as they are created dynamically from pre-existing Objects/Classes. I found a way of doing this by using the declarative Bases class registry:

    def _get_class_from_registry(self, class_name: str) -> None | Type[DeclarativeBase]:
        class_list = [
            c
            for n, c in self.db_base_cls.registry._class_registry.items()
            if n == class_name
        ]
        length = len(class_list)
        assert length in [0, 1]
        if length == 1:
            return class_list.pop()
        if length == 0:
            return None

But I am not satisfied by this solution as I am accessing the protected attribute _class_registry. As an alternative I looked into reflecting my ORM class definition from the database itself with this mock program:

import os

from sqlalchemy import Column, String, create_engine
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session

os.remove("my.db")
Base = automap_base()


class User(Base):
    __tablename__ = "user"

    name = Column("name", String, primary_key=True)


engine = create_engine("sqlite+pysqlite:///my.db", echo=True)
Base.prepare(autoload_with=engine, reflect=True)
Base.metadata.create_all(engine)

user_cls = Base.classes.User
with Session(engine) as session:
    session.add(User(name="test1"))
    session.flush()
    session.commit()

with Session(engine) as session:
    u1 = session.query(User).first()
    print(u1.name)

with Session(engine) as session:
    u2 = session.query(user_cls).first()
    print(u2.name)

But the Base.classes object does not have an attribute User, presumably it is not reflected as it was declared.

My question is how do I implement the former or latter approach correctly?

Answer 1

You can access the mapped classes via the registry's mappers attribute, without having to use a private attribute.

mappers = Base.registry.mappers
class_list = [mapper.entity for mapper in mappers if mapper.entity.__name__ == 'User']