Contenttypes¶
Intro¶
Relational database systems work with the concept of tables. Tables are independent of each other except for foreign keys which works nice in most cases but this design has a little drawback.
Querying, iterating generically across tables and domains is hard, this is where ContentTypes come in play. ContentTypes abstract all the tables in one table, which is quite powerful. By having only one table with back links to all the other tables, it is possible to have generic tables which logic applies to all other tables. Normally you can only enforce uniqueness per table, now it this possible via the ContentType table for data in different tables (you just have to compress them usefully e.g. by a hash).
import edgy
database = edgy.Database("sqlite:///db.sqlite")
models = edgy.Registry(database=database, with_content_type=True)
class Person(edgy.Model):
first_name = edgy.fields.CharField(max_length=100)
last_name = edgy.fields.CharField(max_length=100)
class Meta:
registry = models
unique_together = [("first_name", "last_name")]
class Organisation(edgy.Model):
name = edgy.fields.CharField(max_length=100, unique=True)
class Meta:
registry = models
class Company(edgy.Model):
name = edgy.fields.CharField(max_length=100, unique=True)
class Meta:
registry = models
async def main():
async with database:
await models.create_all()
person = await Person.query.create(first_name="John", last_name="Doe")
org = await Organisation.query.create(name="Edgy org")
comp = await Company.query.create(name="Edgy inc")
# we all have the content_type attribute and are queryable
assert await models.content_type.query.count() == 3
edgy.run_sync(main())
Implementation
Because we allow all sorts of primary keys we have to inject an unique field in every model to traverse back.
Example: The art of comparing apples with pears¶
Let's imagine we have to compare apples with pears via weight. We want only fruits with different weights. Because weight is a small number we just can put it in the collision_key field of ContentType.
import asyncio
from sqlalchemy.exc import IntegrityError
import edgy
database = edgy.Database("sqlite:///db.sqlite")
models = edgy.Registry(database=database, with_content_type=True)
class Apple(edgy.Model):
g = edgy.fields.SmallIntegerField()
class Meta:
registry = models
class Pear(edgy.Model):
g = edgy.fields.SmallIntegerField()
class Meta:
registry = models
async def main():
async with database:
await models.create_all()
await Apple.query.bulk_create(
[{"g": i, "content_type": {"collision_key": str(i)}} for i in range(1, 100, 10)]
)
apples = [
await asyncio.create_task(content_type.get_instance())
async for content_type in models.content_type.query.filter(name="Apple")
]
try:
await Pear.query.bulk_create(
[{"g": i, "content_type": {"collision_key": str(i)}} for i in range(1, 100, 10)]
)
except IntegrityError:
pass
pears = [
await asyncio.create_task(content_type.get_instance())
async for content_type in models.content_type.query.filter(name="Pear")
]
assert len(pears) == 0
edgy.run_sync(main())
If we know we compare over all domains just weight, we can even replace the collision_key field via an IntegerField.
import asyncio
from sqlalchemy.exc import IntegrityError
import edgy
from edgy import Database, Registry, run_sync
from edgy.contrib.contenttypes import ContentType as _ContentType
database = Database("sqlite:///db.sqlite")
class ContentType(_ContentType):
collision_key = edgy.IntegerField(null=True, unique=True)
class Meta:
abstract = True
models = Registry(database=database, with_content_type=ContentType)
class Apple(edgy.Model):
g = edgy.fields.SmallIntegerField()
class Meta:
registry = models
class Pear(edgy.Model):
g = edgy.fields.SmallIntegerField()
class Meta:
registry = models
async def main():
async with database:
await models.create_all()
await Apple.query.bulk_create(
[{"g": i, "content_type": {"collision_key": i}} for i in range(1, 100, 10)]
)
apples = [
await asyncio.create_task(content_type.get_instance())
async for content_type in models.content_type.query.filter(name="Apple")
]
try:
await Pear.query.bulk_create(
[{"g": i, "content_type": {"collision_key": i}} for i in range(1, 100, 10)]
)
except IntegrityError:
pass
pears = [
await asyncio.create_task(content_type.get_instance())
async for content_type in models.content_type.query.filter(name="Pear")
]
assert len(pears) == 0
run_sync(main())
Or now we allow fruits with the same weight. Let's just remove the uniqueness from the collision_key field.
import asyncio
from sqlalchemy.exc import IntegrityError
import edgy
from edgy import Database, Registry, run_sync
from edgy.contrib.contenttypes import ContentType as _ContentType
database = Database("sqlite:///db.sqlite")
class ContentType(_ContentType):
collision_key = edgy.IntegerField(null=True, unique=False)
class Meta:
abstract = True
models = Registry(database=database, with_content_type=ContentType)
class Apple(edgy.Model):
g = edgy.fields.SmallIntegerField()
class Meta:
registry = models
class Pear(edgy.Model):
g = edgy.fields.SmallIntegerField()
class Meta:
registry = models
async def main():
async with database:
await models.create_all()
await Apple.query.bulk_create(
[{"g": i, "content_type": {"collision_key": i}} for i in range(1, 100, 10)]
)
apples = [
await asyncio.create_task(content_type.get_instance())
async for content_type in models.content_type.query.filter(name="Apple")
]
await Pear.query.bulk_create(
[{"g": i, "content_type": {"collision_key": i}} for i in range(1, 100, 10)]
)
pears = [
await asyncio.create_task(content_type.get_instance())
async for content_type in models.content_type.query.filter(name="Pear")
]
run_sync(main())
Example 2: Snapshotting¶
Sometime you want to know when an object is created (or updated), so you can reduce the search area or mark old data for deletion.
Edgy is able to do this quite easily:
import asyncio
from datetime import datetime, timedelta
from sqlalchemy.exc import IntegrityError
import edgy
from edgy import Database, Registry, run_sync
from edgy.contrib.contenttypes import ContentType as _ContentType
database = Database("sqlite:///db.sqlite")
class ContentType(_ContentType):
# Because of sqlite we need no_constraint for the virtual deletion
no_constraint = True
created = edgy.fields.DateTimeField(auto_now_add=True, read_only=False)
keep_until = edgy.fields.DateTimeField(null=True)
class Meta:
abstract = True
models = Registry(database=database, with_content_type=ContentType)
class Person(edgy.Model):
first_name = edgy.fields.CharField(max_length=100)
last_name = edgy.fields.CharField(max_length=100)
class Meta:
registry = models
unique_together = [("first_name", "last_name")]
class Organisation(edgy.Model):
name = edgy.fields.CharField(max_length=100, unique=True)
class Meta:
registry = models
class Company(edgy.Model):
name = edgy.fields.CharField(max_length=100, unique=True)
class Meta:
registry = models
class Account(edgy.Model):
owner = edgy.fields.ForeignKey("ContentType", on_delete=edgy.CASCADE)
class Meta:
registry = models
class Contract(edgy.Model):
owner = edgy.fields.ForeignKey("ContentType", on_delete=edgy.CASCADE)
account = edgy.fields.ForeignKey("Account", null=True, on_delete=edgy.SET_NULL)
class Meta:
registry = models
async def main():
async with database:
await models.create_all()
person = await Person.query.create(first_name="John", last_name="Doe")
snapshot_datetime = datetime.now()
keep_until = snapshot_datetime + timedelta(days=50)
org = await Organisation.query.create(
name="Edgy org",
)
comp = await Company.query.create(
name="Edgy inc",
)
account_person = await Account.query.create(
owner=person.content_type,
content_type={"created": snapshot_datetime, "keep_until": keep_until},
)
account_org = await Account.query.create(
owner=org.content_type,
content_type={"created": snapshot_datetime, "keep_until": keep_until},
contracts_set=[
{
"owner": org.content_type,
"content_type": {"created": snapshot_datetime, "keep_until": keep_until},
}
],
)
account_comp = await Account.query.create(
owner=comp.content_type,
content_type={"created": snapshot_datetime, "keep_until": keep_until},
contracts_set=[
{
"owner": comp.content_type,
"content_type": {"created": snapshot_datetime},
}
],
)
# delete old data
print(
"deletions:",
await models.content_type.query.filter(keep_until__lte=keep_until).delete(),
)
print("Remaining accounts:", await Account.query.count()) # should be 0
print("Remaining contracts:", await Contract.query.count()) # should be 1
print("Accounts:", await Account.query.get_or_none(id=account_comp.id))
edgy.run_sync(main())
Tricks¶
CASCADE deletion does not work or constraint problems¶
Sometime CASCADE deletion is not possible because of the underlying database technology (see snapshotting example) or constraints doesn't work like expected, e.g. slowdown.
You can switch to the virtual CASCADE deletion handling without a constraint by using no_constraint = True.
If you want a completely different handling for one Model, you can use the ContentTypeField and overwrite all extras.
Using in libraries¶
ContentType is always available under the name ContentType if activated and as a content_type attribute on registry.
If the attribute content_type on registry is not None, you can be assured ContentType is available.
Opting out¶
Some models may should not be referencable by ContentType.
You can opt out by overwriting content_type on the model to opt out with any Field.
Use ExcludeField to remove the field entirely.
Tenancy compatibility¶
ContentType is out of the box tenancy compatible.