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 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.
"""
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.
-
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):
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(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.