Tenancy¶

In scenarios where you need to query data from different databases or schemas, Edgy provides robust support for multi-tenancy. This is useful when data is segregated for different users or purposes.

What is Multi-Tenancy?¶

Multi-tenancy architectures commonly fall into three categories:

Shared Schemas: All user data resides in a single schema, differentiated by unique IDs. This approach may have limitations with data privacy regulations like GDPR.
Shared Database, Different Schemas: User data is separated into distinct schemas within the same database.
Different Databases: User data is stored in completely separate databases.

Edgy offers three primary methods for implementing multi-tenancy:

Using in the queryset.
Using with Database in the queryset.
With Tenant for global tenant context.

Edgy also provides helpers for managing schemas, as described in the Schemas section of the Registry documentation.

Schema Creation¶

Edgy simplifies schema creation with the create_schema utility function:

from edgy.core.tenancy.utils import create_schema

This function allows you to create schemas based on a given registry, enabling schema creation across different databases.

Parameters¶

registry: An instance of a Registry.
schema_name: The name of the new schema.
models: An optional dictionary mapping Edgy model names to model classes. If not provided, tables are generated from the registry's models.
if_not_exists: If True, the schema is created only if it doesn't exist. Defaults to False.
should_create_tables: If True, tables are created within the new schema. Defaults to False.

Example:

import edgy
from edgy.core.tenancy.utils import create_schema

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

# Create the schema
await create_schema(
    registry=registry,
    schema_name="edgy",
    if_not_exists=True,
    should_create_tables=True
)

Using¶

The using method allows you to specify a schema for a specific query, overriding the default schema set in the registry.

Parameters:

schema: A string representing the schema name.

Syntax:

<Model>.query.using(schema=<SCHEMA-NAME>).all()

This method can be used with any query type, not just all().

Example:

Consider two schemas, default and other, each containing a User table.

import edgy
from edgy import Database, Registry

database = Database("<YOUR-CONNECTION-STRING>")
models = Registry(database=database)


class User(edgy.Model):
    is_active: bool = edgy.BooleanField(default=False)

    class Meta:
        registry = models

Querying the default schema:

User.query.all()

Querying the main schema:

User.query.using(schema='main').all()

Using with Database¶

The using_with_db method allows you to query a schema in a different database.

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.

Set Tenant¶

The with_tenant context manager sets a global tenant context for your application, ensuring all queries within the context target the specified tenant's data.

from edgy.core.db import with_tenant

Tip

Use with_tenant in middleware or interceptors to set the tenant context before API requests.

Warning

The set_tenant function is deprecated and should not be used.

Practical Case¶

Let's illustrate with_tenant with an Esmerald application example.

Building:

Models: Define models for tenants, users, and products.
Middleware: Intercept requests and set the tenant context.
API: Create an API to retrieve tenant-specific data.

Models¶

Define models to represent tenants, users, and products:

from typing import Any, Dict, List, Type, Union

import sqlalchemy
from loguru import logger

import edgy
from edgy.core.db.models.model import Model
from edgy.core.db.models.utils import get_model
from edgy.testclient import DatabaseTestClient as Database

database = Database("<YOUR-CONNECTION-STRING>")
registry = edgy.Registry(database=database)


class Tenant(edgy.Model):
    schema_name: str = edgy.CharField(max_length=63, unique=True, index=True)
    tenant_name: str = edgy.CharField(max_length=100, unique=True, null=False)

    class Meta:
        registry = registry

    def table_schema(
        self,
        model_class: Type["Model"],
        schema: str,
    ) -> sqlalchemy.Table:
        return model_class.table_schema(schema)

    async def create_tables(
        self,
        registry: "edgy.Registry",
        tenant_models: Dict[str, Type["Model"]],
        schema: str,
        exclude: Union[List[str], None],
    ) -> None:
        for name, model in tenant_models.items():
            if name in exclude:
                continue

            table = self.table_schema(model, schema)

            logger.info(f"Creating table '{model.meta.tablename}' for schema: '{schema}'")
            try:
                async with registry.engine.begin() as connection:
                    await connection.run_sync(table.create)
                await registry.engine.dispose()
            except Exception as e:
                logger.error(str(e))

    async def create_schema(self) -> None:
        await registry.schema.create_schema(self.schema_name)

    async def save(
        self,
        force_insert: bool = False,
        values: Union[dict[str, Any], set[str], None] = None,
    ) -> Type[Model]:
        tenant = await super().save(force_insert, values)
        try:
            await self.create_schema(schema=tenant.schema_name, if_not_exists=True)
            await self.create_tables(
                registry, registry.models, tenant.schema_name, exclude=["Tenant", "TenantUser"]
            )
        except Exception as e:
            message = f"Rolling back... {str(e)}"
            logger.error(message)
            await self.delete()
        return tenant


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

    class Meta:
        registry = registry


class Product(edgy.Model):
    name: str = edgy.CharField(max_length=255)
    user: User = edgy.ForeignKey(User, null=True)

    class Meta:
        registry = registry

The TenantUser model links database schemas to users.

Generate Example Data¶

Populate the tables with example data:

async def create_data():
    """
    Creates mock data.
    """
    # Create some users in the main users table
    esmerald = await User.query.create(name="esmerald")

    # Create a tenant for Edgy (only)
    tenant = await Tenant.query.create(
        schema_name="edgy",
        tenant_name="edgy",
    )

    # Create a user in the `User` table inside the `edgy` tenant.
    edgy = await User.query.using(schema=tenant.schema_name).create(
        name="Edgy schema user",
    )

    # Products for Edgy (inside edgy schema)
    for i in range(10):
        await Product.query.using(schema=tenant.schema_name).create(
            name=f"Product-{i}",
            user=edgy,
        )

    # Products for Esmerald (no schema associated, defaulting to the public schema or "shared")
    for i in range(25):
        await Product.query.create(name=f"Product-{i}", user=esmerald)

Middleware¶

Create middleware to intercept requests and set the tenant context:

from typing import Any, Coroutine

from esmerald import Request
from esmerald.protocols.middleware import MiddlewareProtocol
from lilya.types import ASGIApp, Receive, Scope, Send

from edgy.core.db import with_tenant
from edgy.exceptions import ObjectNotFound


class TenantMiddleware(MiddlewareProtocol):
    def __init__(self, app: "ASGIApp"):
        super().__init__(app)
        self.app = app

    async def __call__(
        self, scope: Scope, receive: Receive, send: Send
    ) -> Coroutine[Any, Any, None]:
        """
        Receives a header with the tenant information and lookup in
        the database if exists.

        Sets the tenant if true, or none otherwise.
        """
        request = Request(scope=scope, receive=receive, send=send)
        tenant_header = request.headers.get("tenant", None)

        try:
            user = await Tenant.query.get(schema_name=tenant_header)
            tenant = user.schema_name
        except ObjectNotFound:
            tenant = None

        with with_tenant(tenant):
            await self.app(scope, receive, send)

The with_tenant context manager sets the tenant context for all API calls.

API¶

Create an Esmerald API to retrieve product data:

from typing import List

from esmerald import Esmerald, Gateway, get

import edgy

database = edgy.Database("<TOUR-CONNECTION-STRING>")
models = edgy.Registry(database=database)


@get("/products")
async def products() -> List[Product]:
    """
    Returns the products associated to a tenant or
    all the "shared" products if tenant is None.
    """
    products = await Product.query.all()
    return products


app = models.asgi(
    Esmerald(
        routes=[Gateway(handler=products)],
        middleware=[TenantMiddleware],
    )
)

Query the API¶

Querying the API should return data corresponding to the specified tenant:

import httpx


async def query():
    response = await httpx.get("/products", headers={"tenant": "edgy"})

    # Total products created for `edgy` schema
    assert len(response.json()) == 10

    # Response for the "shared", no tenant associated.
    response = await httpx.get("/products")
    assert len(response.json()) == 25

Tenant Only Models¶

To prevent models from being created in the non-tenant schema, set register_default to False in the model's Meta.

Notes¶

The with_tenant context manager is particularly useful for large-scale multi-tenant applications, simplifying tenant data management.