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

    id: int = edgy.IntegerField(primary_key=True)
    is_active: bool = edgy.BooleanField(default=False)

    class Meta:
        registry = models

Parameters¶

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

Warning

Using the Database from the databases package will raise an assertation error. You must use the edgy.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 edgy 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 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.
    """

    id: int = edgy.IntegerField(primary_key=True)
    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.

^{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}

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):
    id: int = fields.IntegerField(primary_key=True)
    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):
    id: int = fields.IntegerField(primary_key=True)
    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_with_db("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. Single sub-components can be excluded.

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

Model class attributes class_attrs which are cleared or set are table, pknames, pkcolumns.

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.

db_schema internally calls also invalidate_models() to remove table references.