Registry¶

When using the Saffier ORM, you must use the Registry object to tell exactly where the database is going to be.

Imagine the registry as a mapping between your models and the database where is going to be written.

And is just that, nothing else and very simple but effective object.

The registry is also the object that you might want to use when generating migrations using Alembic.

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

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")
```

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

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": "Edgy", "email": "saffier@example.com"},
            {"name": "Edgy 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.