Registry

When working with the Edgy ORM, the Registry object is essential for specifying the database connection.

Think of the registry as a mapping between your models and the database where data will be stored.

It's a simple yet effective object with a crucial role. The registry is also used for generating migrations with Alembic.

import edgy
from edgy import Database, Registry

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


class User(edgy.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.
    """

    is_active: bool = edgy.BooleanField(default=False)

    class Meta:
        registry = models

Parameters

• database: An instance of edgy.core.db.Database or a connection string. When using a string, unparsed keyword arguments are passed to the created Database object.

  Warning

  Using Database from the databases package raises an assertion error. Edgy uses the databasez fork, and it's recommended to use a string, edgy.Database, or edgy.testclient.TestClient. Future versions may add more Edgy-specific functionality.

• schema: The schema to connect to. Useful for multi-tenancy applications or connecting to non-default schemas.

  from edgy import Registry
  
  registry = Registry(database=..., schema="custom-schema")
  

• extra: A dictionary of extra connections (same types as the database argument) managed by the registry (connecting/disconnecting). They can be arbitrary connected databases. It ensures they're not torn down while the registry is connected.

• with_content_type: A boolean or a custom abstract ContentType prototype. Enables ContentTypes and saves the used type as the content_type attribute.

Connecting/Disconnecting

Registries support the asynchronous context manager protocol and the ASGI lifespan protocol. This ensures all databases specified in database or extra are properly referenced and dereferenced (triggering initialization and teardown when the reference count reaches 0). This allows safe use of databases across different contexts.

Accessing ContentType

The registry has a content_type attribute for accessing the active ContentType.

Direct Database Access

The registry has a database attribute for the main database and an extra dictionary for extra databases. Retrieving the Database object from the registry is safer and ensures you get the correct instance.

Custom Registry

You can create custom registries by subclassing the Registry class.

import edgy
from edgy 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(edgy.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.
    """

    is_active: bool = edgy.BooleanField(default=False)

    class Meta:
        registry = models

Multiple Registries

You can work with multiple databases across different functionalities using multiple registries with Meta combinations.

import edgy
from edgy 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(edgy.Model):
    is_active: bool = edgy.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(edgy.Model):
    is_active: bool = edgy.BooleanField(default=False)

    class Meta:
        registry = another_registry

Schemas

Edgy supports database schema operations like creating schemas and dropping schemas.

This is useful for multi-tenancy applications or custom schema management.

Create Schema

Creates database schemas.

Parameters:

• schema: String name of the schema.
• if_not_exists: Flag to create if the schema doesn't exist.

• databases: String or None for the main database. You can create schemas on databases in extra too.

  ^{Default: False}

from edgy 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 named edgy.

await create_schema("edgy")

This creates the edgy schema if it doesn't exist. If if_not_exists is False and the schema exists, it raises edgy.exceptions.SchemaError.

Drop Schema

Drops database schemas.

Warning

Use drop_schema with caution, as it's irreversible. Avoid deleting the default schema.

Parameters:

• schema: String name of the schema.

• cascade: Flag for cascade delete.

  ^{Default: False}

• if_exists: Flag to drop if the schema exists.

  ^{Default: False} * databases: String or None for main database. You can drop schemes on databases in extra too.

from edgy 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 named edgy.

await drop_schema("edgy")

This drops the edgy schema if it exists. If if_exists is False and the schema doesn't exist, it raises edgy.exceptions.SchemaError.

Get Default Schema Name

Helper function to get the default schema name for the database (e.g., public for Postgres, dbo for MSSQL).

from edgy 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 Edgy 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 edgy
from edgy.core.db import fields
from edgy.testclient import DatabaseTestClient as Database

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


class User(edgy.Model):
    name: str = fields.CharField(max_length=255)
    email: str = 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 edgy
from edgy.core.db import fields
from edgy.testclient import DatabaseTestClient as Database

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


class User(edgy.Model):
    name: str = fields.CharField(max_length=255)
    email: str = fields.CharField(max_length=255)

    class Meta:
        registry = models


async def bulk_create_users() -> None:
    """
    Bulk creates some users.
    """
    await User.query.using(database="alternative").bulk_create(
        [
            {"name": "Edgy", "email": "edgy@example.com"},
            {"name": "Edgy Alternative", "email": "edgy.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.

Laziness

For advanced users who want to control the laziness of meta objects.

Registry objects have helper functions to undo laziness (for optimizations or static environments):

• init_models(self, *, init_column_mappers=True, init_class_attrs=True): Fully initializes models and metas. Exclude elements by setting keyword arguments to False.
• invalidate_models(self, *, clear_class_attrs=True): Invalidates metas and removes cached class attributes. Exclude sub-components from invalidation.

Model class attributes (table, pknames, pkcolumns, proxy_model, table_schema) are cleared or initialized.

Manual initialization is usually unnecessary and can cause performance penalties.

init_column_mappers initializes columns_to_field via its init() method, which can be expensive for large models.

Callbacks

Use callbacks to modify models or specific models when they're available.

Register callbacks with a model name or None (for all models). When a model class is added, the callback is executed with the model class as a parameter.

Callbacks can be permanent or one-time (triggered by the first match). If a model is already registered, it's passed to the callback.

Use register_callback(model_or_name, callback, one_time).

Generally, use one_time=True for model-specific callbacks and one_time=False for model-unspecific callbacks.

If one_time is not provided, the logic mentioned above is applied.