Automatic Reflection¶

Let's reflect reflection:

we can reflect tables from database in a model. The next step is to retrieve the tables and create reflection models from it automatically. This can be useful to create interfaces procedural.

For doing so we have selections via Pattern Models:

They contain a Meta with the regex attribute and generate via the template string or function a new ReflectionModel:

from edgy.contrib.autoreflection import AutoReflectModel

class Reflected(AutoReflectModel):
    class Meta:
        include_pattern = ".*"  # regex or string, default .*. Matches against the tablename
        exclude_pattern = None  # regex or string, default None (disabled). Matches against the tablename
        template = "{modelname}{tablename}"  # string or function with arguments tablename, modelname, tablekey
        databases = (None,)  # Restrict reflection to databases. None: main database of registry, string extra databases of registry
        schemes = (None,) # Which schemes to checks

Note: when a reflected model is generated the meta is switched in the copy to a regular MetaInfo.

Meta parameters¶

Note: we use the table name not the table key.

What is the difference:

table name: tablename table key: schema.tablename

That is because edgy handle schemes a little bit different:

In edgy a model can exist in multiple schemes and the scheme is explicit selected.

Inclusion & Exclusion patterns¶

Here we can specify patterns against which a regex is checked. By default the include_pattern is set to .* which matches all tablenames and the exclude_pattern is disabled.

The include_pattern will convert all falsy values to the match all, while the exclude_pattern will be really disabled.

Template¶

Can be a function which takes the tablename as parameter and is required to return a string.

Or it can be a format string with the possible parameters:

tablename: the name of the table
tablekey: the key (name with scheme) of the table
modelname: the model name

Databases¶

In the registry you specify a main database (which is here None) and via the extra dictionary multiple named databases. The extra databases can be selected via their name while the main can be selected by None.

This controls from which database the models are reflected. This is useful to extract data from other databases and to use it in the main application.

By default the autoreflection only uses the main databases.

Schemes¶

This parameter is providing the schemes which should be scanned for models.

This parameter is required when the models which should be reflected are in a different schema.

Examples¶

Procedural interface¶

To build an application there is also a data driven approach. Instead of defining relations and fields by hand they are all automatically generated.

For creating the tables we can use:

source.py

import edgy

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


class Trouser(edgy.Model):
    price = edgy.DecimalField(max_digits=2, decimal_places=2)
    name = edgy.CharField(max_length=50)
    with_pocket = edgy.BooleanField()
    size = edgy.IntegerField()

    class Meta:
        registry = registry


class Shoe(edgy.Model):
    price = edgy.DecimalField(decimal_places=2)
    name = edgy.CharField(max_length=50)
    waterproof = edgy.BooleanField()
    size = edgy.FloatField()

    class Meta:
        registry = registry


async def main():
    async with registry:
        await registry.create_all()
        await Trouser.query.create(price=10.50, name="Fancy Jeans", with_pocket=True, size=30)
        await Shoe.query.create(price=14.50, name="SuperEliteWalk", waterproof=False, size=10.5)


edgy.run_sync(main())

Then we can reflect:

procedural.py

import edgy
from edgy.contrib.autoreflection import AutoReflectModel

reflected = edgy.Registry(database="sqlite:///webshopdb.sqlite")


class Product(AutoReflectModel):
    price = edgy.DecimalField(decimal_places=2)
    name = edgy.CharField(max_length=50)

    class Meta:
        registry = reflected
        template = r"{modelname}_{tablename}"


async def main():
    async with reflected:
        print(
            *[
                product.model_dump()
                async for product in reflected.get_model("Product_shoes").query.all()
            ]
        )
        print(
            *[
                product.model_dump()
                async for product in reflected.get_model("Product_trousers").query.all()
            ]
        )
        await reflected.get_model("Product_shoes").query.update(
            price=reflected.get_model("Product_shoes").table.c.price + 10
        )


edgy.run_sync(main())

Legacy databases¶

Suppose you have a new modern database, a legacy database and an ancient database which very few capabilities from which both you need data. In the legacy and ancient database, you are only allowed to update some specific fields.

legacy.py

import edgy
from edgy.contrib.autoreflection import AutoReflectModel

registry = edgy.Registry(
    database="sqlite:///newapp.sqlite",
    extra={
        "legacy": "sqlite:///webshopdb.sqlite",
        "ancient": "sqlite:///webshopdb.sqlite",
    },
)


class Product(AutoReflectModel):
    price = edgy.DecimalField(decimal_places=2)
    name = edgy.CharField(max_length=50)

    class Meta:
        registry = registry
        template = r"{modelname}_{tablename}"
        databases = ("legacy",)


class AncientProduct(edgy.ReflectModel):
    price = edgy.DecimalField(decimal_places=2)
    name = edgy.CharField(max_length=50)
    __using_schema__ = None

    class Meta:
        registry = registry
        tablename = "shoes"


AncientProduct.database = registry.extra["ancient"]


async def main():
    async with registry:
        print(*[product.model_dump() async for product in AncientProduct.query.all()])
        print(
            *[
                product.model_dump()
                async for product in registry.get_model("Product_shoes").query.all()
            ]
        )
        print(
            *[
                product.model_dump()
                async for product in registry.get_model("Product_trousers").query.all()
            ]
        )
        await registry.get_model("Product_shoes").query.update(
            price=registry.get_model("Product_shoes").table.c.price + 10
        )


edgy.run_sync(main())