SqlAlchemy Registry

After talking about Persistence Ignorance and mapping styles in SqlAlchemy in my previous post, it's time now to take a look to an interesting technique used by the Declarative mapping. Whatever mapping style you use, SqlAlchemy relies on a registry where the information that maps entities to tables, fields to properties, relations, etc is stored. When using the Imperative mapping you directly work with that registry (sqlalchemy.orm.registry)

from sqlalchemy.orm import registry
@dataclasses.dataclass
class Post:
    title: int
    content: str

metadata = MetaData()
mapper_registry = registry(metadata=metadata)

mapper_registry.map_imperatively(
	entities.Post,
	table_post,
	properties={
		#"post_id": table_post.c.PostId, 
		"title": table_post.c.Title,
		"content": table_post.c.Content
	}      
)

But when using the declarative mapping you're not aware of that registry as you normally don't interact with it at all, though notice that you still have access to it through the Base class.

class Base(DeclarativeBase):
    pass
	
class Post(Base):
    __tablename__ = "Posts"
    post_id: Mapped[int] = mapped_column("PostId", primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column("Title")
    content: Mapped[str] = mapped_column("Content")
	
# we have not directly used the registry at all in the above code, but it's still there, accessible through the Base class:
print(f"{Base.registry=}")
# Base.registry=
    

So how does the registry get set? Well, your entities get registered in that registry by leveraging the inheritance and metaclasses machinery to obtain a behaviour that is similar to the ruby inherited hook. Remember that I already talked in a previous post about simulating another ruby metaprogramming hook, the method_added hook, by means of metaclasses. We can use metaclasses to execute some action each time a class based on that metaclass is created (putting that code to be executed in the __new__ or __init__ methods of the metaclass). In our case, we want to execute code to add each model class to the registry. For that the Base class that we define for our entities must have DeclarativeMeta as its metaclass. We can do this by directly setting ourselves the metaclass and the registry instance:

mapper_registry = registry()

class Base(metaclass=DeclarativeMeta):
    registry = mapper_registry

Or by inheriting from DeclarativeBase (that already has DeclarativeMeta as its meta). DeclarativeBase will also take care of setting the registry in our Base class.

class Base(DeclarativeBase):
    pass

We can take look at the DeclarativeMeta code to see how it makes its magic:

class DeclarativeMeta(DeclarativeAttributeIntercept):
    metadata: MetaData
    registry: RegistryType

    def __init__(
        cls, classname: Any, bases: Any, dict_: Any, **kw: Any
    ) -> None:
        # use cls.__dict__, which can be modified by an
        # __init_subclass__() method (#7900)
        dict_ = cls.__dict__

        # early-consume registry from the initial declarative base,
        # assign privately to not conflict with subclass attributes named
        # "registry"
        reg = getattr(cls, "_sa_registry", None)
        if reg is None:
            reg = dict_.get("registry", None)
            if not isinstance(reg, registry):
                raise exc.InvalidRequestError(
                    "Declarative base class has no 'registry' attribute, "
                    "or registry is not a sqlalchemy.orm.registry() object"
                )
            else:
                cls._sa_registry = reg

        if not cls.__dict__.get("__abstract__", False):
            _ORMClassConfigurator._as_declarative(reg, cls, dict_)
        type.__init__(cls, classname, bases, dict_)
from sqlalchemy.orm import DeclarativeBase

I'm involved in some projects where the Database is not a critical element. We don't retrieve data from it, we just use it as an additional storage for our results, but the main store for those results are json/csv files. This means that if the Database is down, the application should run anyway. So it's important for me to have clear what things involve database access (and hence an error if the DB is not accessible), and also when the model mapping will throw an error if the mapping is incorrect. Let's see:

• Adding classes to the registry (either explicitly with the imperative mapping or implicitly with the declarative one) does not perform any check with the database (so if the DB is down or there's something wrong in our mapping, like wrong tables or columns, we won't find it until later).
• Creating a SqlAlchemy engine does not perform any connection to the DB either.
• Creating a Session does connect to the Database, but it does not perform any model verification.
• Adding objects to a Session won't check the model until the moment when you do a flush or a commit (that indirectly performs a flush).
• Performing a select Query through a Session will obviously generate an error if any of the mappings for the tables involved in the query is wrong.

invoke operator function