Registry¶

The Registry is the runtime hub of a Saffier application.

Models point at a registry, querysets resolve tables and metadata through it, reflection code stores reflected models inside it, and migration helpers use it as the source of truth for generated SQLAlchemy metadata.

If you only remember one thing, remember this:

models describe structure, the registry describes where that structure lives.

import saffier
from saffier import Database, Registry

database = Database("sqlite:///db.sqlite")
models = Registry(database=database)


class User(saffier.Model):
    """
    The User model to be created in the database as a table
    If no name is provided the in Meta class, it will generate
    a "users" table for you.
    """

    id = saffier.IntegerField(primary_key=True)
    is_active = saffier.BooleanField(default=False)

    class Meta:
        registry = models

Mental model¶

A registry owns:

the primary Database
any extra= databases
registered models
reflected models
SQLAlchemy metadata for each configured database

That is why it appears in model declarations, migration setup, reflection, and multi-database applications.

Parameters¶

database - An instance of saffier.core.db.Database object.

Warning

Using the Database from the databases package will raise an assertation error. You must use the saffier.Database object instead.

schema - The schema to connect to. This can be very useful for multi-tenancy applications if you want to specify a specific schema or simply if you just want to connect to a different schema that is not the default.
```
from saffier import Registry

registry = Registry(database=..., schema="custom-schema")
```
model_engine - Optional default engine adapter used by models in this registry.
```
registry = Registry(database=..., model_engine="pydantic")
```

Practical single-registry setup¶

Most applications only need one registry:

database = saffier.Database(
    "postgresql+asyncpg://postgres:postgres@localhost:5432/app"
)
models = saffier.Registry(database=database)

Then every model declares registry = models inside Meta.

Registry-wide model engine defaults¶

If most models in one application should expose the same engine-backed projection, configure it once on the registry:

models = saffier.Registry(database=database, model_engine="pydantic")

Models then inherit that default unless they override it with Meta.model_engine or opt out via Meta.model_engine = False.

This keeps engine selection at the same application boundary where database, metadata, and model registration are already configured.

The built-in adapter names are currently pydantic and msgspec, but custom engines can also be registered.

Custom registry¶

Can you have your own custom Registry? Yes, of course! You simply need to subclass the Registry class and continue from there like any other python class.

import saffier
from saffier import Database, Registry


class MyRegistry(Registry):
    """
    Add logic unique to your registry or override
    existing functionality.
    """

    ...


database = Database("sqlite:///db.sqlite")
models = MyRegistry(database=database)


class User(saffier.Model):
    """
    The User model to be created in the database as a table
    If no name is provided the in Meta class, it will generate
    a "users" table for you.
    """

    id = saffier.IntegerField(primary_key=True)
    is_active = saffier.BooleanField(default=False)

    class Meta:
        registry = models

Multiple registries¶

Sometimes you might want to work with multiple databases across different functionalities and that is also possible thanks to the registry with Meta combination.

import saffier
from saffier import Database, Registry


class MyRegistry(Registry):
    """
    Add logic unique to your registry or override
    existing functionality.
    """

    ...


database = Database("sqlite:///db.sqlite")
models = MyRegistry(database=database)


class User(saffier.Model):
    is_active = saffier.BooleanField(default=False)

    class Meta:
        registry = models


another_db = Database("postgressql://user:password@localhost:5432/mydb")
another_registry = MyRegistry(another_db=another_db)


class Profile(saffier.Model):
    is_active = saffier.BooleanField(default=False)

    class Meta:
        registry = another_registry

When you use the extra= mapping for multiple databases, keep the keys stable and string-based. Saffier preserves a separate SQLAlchemy MetaData object per configured database and exposes both metadata_by_name and metadata_by_url for migration/runtime code that needs to target a specific connection.

Names with surrounding whitespace still work, but Saffier warns about them because they make CLI and migration output harder to read.

Copying a Registry¶

Migration workflows sometimes need an isolated copy of the registry instead of mutating the active application registry in place.

import copy

import saffier


registry_copy = copy.copy(models)
prepared_copy = saffier.get_migration_prepared_registry(registry_copy)

The copied registry keeps:

per-database metadata mappings
reflected-model metadata refresh behavior
copied many-to-many through models wired to the copied registry

That makes it safe to prepare migration metadata without leaving copied relationship state attached to the original registry.

Dynamic model registration¶

Saffier models can also be copied and attached to another registry at runtime.

user_copy = User.copy_saffier_model()
tenant_user = user_copy.add_to_registry(tenant_registry)

This is useful for migration preparation, extension-driven model composition, and tests that need isolated registries. If a copied model still references models that are added later, Saffier now defers the reverse-relation wiring until those models are registered in the target registry.

Automigration¶

For managed runtimes, a registry can also opt into running migrations on first connect:

from saffier import Registry
from myproject.configs.settings import Settings


registry = Registry(database=database, automigrate_config=Settings)

This delegates to the normal migration upgrade() flow and is still controlled by settings.allow_automigrations.

Synchronous Registry Environments¶

When you need to drive async registry lifecycle and ORM calls from synchronous code, use Registry.with_async_env():

import saffier


with models.with_async_env():
    saffier.run_sync(models.create_all())
    user = saffier.run_sync(User.query.create(name="Saffier"))
    fetched = saffier.run_sync(User.query.get(pk=user.pk))

This binds run_sync() to the registry-managed event loop for the duration of the context and cleans the registry up on exit. Nested with_async_env() blocks are supported.

Schemas¶

This is another great supported feature from Saffier. This allows you to manipulate database schema operations like creating schemas or dropping schemas.

This can be particulary useful if you want to create a multi-tenancy application and you need to generate schemas for your own purposes.

Create schema¶

As the name suggests, it is the functionality that allows you to create database schemas.

Parameters:

schema - String name of the schema.
if_not_exists - Flag indicating if should create if not exists.

^{Default: False}

from saffier import Database, Registry

database = Database("<YOUR-CONNECTION-STRING>")
registry = Registry(database=database)


async def create_schema(name: str) -> None:
    """
    Creates a new schema in the database.
    """
    await registry.schema.create_schema(name, if_not_exists=True)

Create a schema called saffier.

await create_schema("saffier")

This will make sure it will create a new schema saffier if it does not exist. If the if_not_exists is False and the schema already exists, it will raise a saffier.exceptions.SchemaError.

Drop schema¶

As name also suggests, it is the opposite of create_schema and instead of creating it will drop it from the database.

Warning

You need to be very careful when using the drop_schema as the consequences are irreversible and not only you don't want to remove the wrong schema but also you don't want to delete the default schema as well. Use it with caution.

Parameters:

schema - String name of the schema.
cascade - Flag indicating if should do cascade delete. * ^{Default: False}
if_exists - Flag indicating if should create if not exists.

^{Default: False}

from saffier import Database, Registry

database = Database("<YOUR-CONNECTION-STRING>")
registry = Registry(database=database)


async def drop_schema(name: str) -> None:
    """
    Drops a schema from the database.
    """
    await registry.schema.drop_schema(name, if_exists=True)

Drop a schema called saffier

await drop_schema("saffier")

This will make sure it will drop a schema saffier if exists. If the if_exists is False and the schema does not exist, it will raise a saffier.exceptions.SchemaError.

Get default schema name¶

This is just a helper. Each database has its own default schema name, for example, Postgres calls it public and MSSQLServer calls it dbo.

This is just an helper in case you need to know the default schema name for any needed purpose of your application.

from saffier import Database, Registry

database = Database("<YOUR-CONNECTION-STRING>")
registry = Registry(database=database)


async def get_default_schema() -> str:
    """
    Returns the default schema name of the given database
    """
    await registry.schema.get_default_schema()

Extra¶

This is the part that makes a whole difference if you are thinking about querying a specific database using a diffent connection.

What does that even mean? Imagine you have a main database public (default) and a database copy somewhere else called alternative (or whatever name you choose) and both have the model User.

You now want to query the alternative to gather some user data that was specifically stored in that database where the connection string is different.

The way Saffier operates is by checking if that alternative connection exists in the extra parameter of the registry and then uses that connection to connect and query to the desired database.

Warning

To use the alternative database, the connection must be declared in the registry of the model or else it will raise an AssertationError.

The way of doing that is by using the using_with_db of the queryset. This is particularly useful if you want to do some tenant applications or simply connecting to a different database to gather your data.

Simple right?

Nothing like a good example to simplify those possible confusing thoughts.

Let us assume we want to bulk_create some users in the alternative database instead of the default.

import saffier
from saffier.core.db import fields
from saffier.testclient import DatabaseTestClient as Database

database = Database("<YOUR-CONNECTION-STRING>")
alternative = Database("<YOUR-ALTERNATIVE-CONNECTION-STRING>")
models = saffier.Registry(database=database, extra={"alternative": alternative})


class User(saffier.Model):
    id = fields.IntegerField(primary_key=True)
    name = fields.CharField(max_length=255)
    email = fields.CharField(max_length=255)

    class Meta:
        registry = models

As you can see, the alternative was declared in the extra parameter of the registry of the model as required.

Now we can simply use that connection and create the data in the alternative database.

import saffier
from saffier.core.db import fields
from saffier.testclient import DatabaseTestClient as Database

database = Database("<YOUR-CONNECTION-STRING>")
alternative = Database("<YOUR-ALTERNATIVE-CONNECTION-STRING>")
models = saffier.Registry(database=database, extra={"alternative": alternative})


class User(saffier.Model):
    id = fields.IntegerField(primary_key=True)
    name = fields.CharField(max_length=255)
    email = fields.CharField(max_length=255)

    class Meta:
        registry = models


async def bulk_create_users() -> None:
    """
    Bulk creates some users.
    """
    await User.query.using_with_db("alternative").bulk_create(
        [
            {"name": "Saffier", "email": "saffier@example.com"},
            {"name": "Saffier Alternative", "email": "saffier.alternative@example.com"},
        ]
    )

Did you notice the alternative name in the using_with_db? Well, that should match the name given in the extra declaration of the registry.

You can have as many connections declared in the extra as you want, there are no limits.