Contenttypes¶
Intro¶
Relational database systems operate using tables that are generally independent, except for foreign keys, which work well in most cases. However, this design has a minor drawback.
Querying and iterating generically across tables and domains can be challenging. This is where ContentTypes come into play. ContentTypes abstract all tables into a single table, providing a powerful solution. By maintaining a single table with backlinks to all other tables, it becomes possible to create generic tables with logic that applies universally.
Typically, uniqueness can only be enforced within individual tables. However, with the ContentType table, it is now possible to enforce uniqueness across multiple tables—this can be achieved efficiently by compressing the data, for example, using 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
Since we allow various types of primary keys, we must inject a unique field into every model to enable backward traversal.
Example: The art of comparing apples with pears¶
Imagine we need to compare apples and pears based on weight, ensuring only fruits with different weights are considered.
Since weight is a small number, we can simply store 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 that the comparison across all domains is based solely on weight, we can even replace the collision_key field with 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, if we now allow fruits with the same weight, we can simply remove the uniqueness constraint 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¶
Sometimes, you may need to track when an object is created or updated to narrow the search scope or mark outdated data for deletion.
Edgy makes this process straightforward:
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 Issues or Constraint Problems¶
Sometimes, CASCADE deletion is not possible due to limitations in the underlying database technology (see the snapshotting example) or unexpected constraint behavior, such as performance slowdowns.
To handle this, you can switch to virtual CASCADE deletion without enforcing a constraint by setting no_constraint = True.
If you need a completely different deletion strategy for a specific model, you can use the ContentTypeField and override all extras.
Using in Libraries¶
If activated, ContentType is always available under the name ContentType and as a content_type attribute on the registry.
If the content_type attribute on the registry is not None, you can be sure that ContentType is available.
Opting Out¶
Some models should not be referencable by ContentType.
To opt out, override content_type on the model with any field. Use ExcludeField to remove the field entirely.
Tenancy Compatibility¶
ContentType is tenancy-compatible out of the box.