Registry

When using the Edgy 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 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 object or a string. When providing a string all unparsed keyword arguments are passed the created Database object.

Warning

Using the Database from the databases package will raise an assertation error. Edgy is build on the fork databasez and it is strongly recommended to use a string, edgy.Database or edgy.testclient.TestClient instead. In future we may add more edgy specific functionality.

• 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 edgy import Registry
  
  registry = Registry(database=..., schema="custom-schema")
  

• extra - A dictionary with extra connections (same types like the database argument) which are managed by the registry too (connecting/disconnecting). They may can be arbitary connected databases. It is just ensured that they are not tore down during the registry is connected.

• with_content_type - Either a bool or a custom abstract ContentType prototype. This enables ContentTypes and saves the actual used type as attribute: content_type

Connecting/Disconnecting

Registries support the async contextmanager protocol as well as the ASGI lifespan protocol. This way all databases specified as database or extra are properly referenced and dereferenced (triggering the initialization and tear down routines when reaching 0). This way all of the dbs can be safely used no matter if they are used in different contexts.

Accessing the ContentType

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

Accessing directly the databases

The registry has an attribute database for the main database and a dictionary extra containing the active extra databases. It is not necessary anymore to keep the Database object available, it can be simply retrieved from the db which is by the way safer. This way it is ensured you get the right one.

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 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

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 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

This is another great supported feature from Edgy. 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.

• databases - String or None for main database. You can create schemes 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 called edgy.

await create_schema("edgy")

This will make sure it will create a new schema edgy if it does not exist. If the if_not_exists is False and the schema already exists, it will raise a edgy.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} * 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 called edgy

await drop_schema("edgy")

This will make sure it will drop a schema edgy if exists. If the if_exists is False and the schema does not exist, it will raise a edgy.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 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.

Lazyness

Note: this is something for really advanced users who want to control the lazyness of meta objects. Skip if you just want use the framework and don't want to micro-optimize your code.

Registry objects have two helper functions which can undo the lazyness (for optimizations or in case of an environment which requires everything being static after init.):

init_models(self, *, init_column_mappers=True, init_class_attrs=True) - Fully initializes models and metas. Some elements can be excluded from initialization by providing False to the keyword argument.

invalidate_models(self, *, clear_class_attrs=True) - Invalidates metas and removes cached class attributes. Single sub-components can be excluded from inval.

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

However in most cases it won't be necessary to initialize them manually and causes performance penalties.

init_column_mappers initializes the columns_to_field via its init() method. This initializes the mappers columns_to_field, field_to_columns and field_to_column_names. This can be expensive for large models.

Callbacks

Sometimes you want to modify all models or a specific model but aren't sure if they are available yet. Here we have now the callbacks in a registry. You register a callback with a model name or None (for all models) and whenever a model class of the criteria is added the callback with the model class as parameter is executed. Callbacks can be registered permanent or one time (the first match triggers them). If a model is already registered it is passed to the callback too. The method is called register_callback(model_or_name, callback, one_time).

Generally you 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.