Skip to content

Model metadata

API Documentation

amsdal_utils.models.data_models.metadata.Metadata

AMSDAL creates metadata for every mutuation of any data it's a specific object that stores some meta information about the record of model. This information is managed by AMSDAL framework, although you can read it and even use it in queries to database.

For example, you can use metadata to query the records that are were deleted:

from models.user.person import Person

deleted_persons = Person.objects.filter(_metadata__is_deleted=True).execute()
from models.user.person import Person

deleted_persons = await Person.objects.filter(_metadata__is_deleted=True).aexecute()

Although, you cannot directly modify the metadata. It's managed by AMSDAL framework itself.

Address

Every record has an address field that stores the information about the object. It's a Address object. It defines particular place of the object in the database, and constists of connection name, class name, class version, object id and object version.

from models.user.person import Person

person = Person(first_name="Jane")
person.save()
person.last_name = "Roe"
person.save()

latest_version, old_version = Person.objects.using(
    LAKEHOUSE_DB_ALIAS,
).order_by('-_metadata__updated_at').execute()


# Two versions of the same object have the same object_id
assert old_version.get_metadata().address.object_id == latest_version.get_metadata().address.object_id
# But different object_versions
assert old_version.get_metadata().address.object_version != latest_version.get_metadata().address.object_version
from models.user.person import Person

person = Person(first_name="Jane")
await person.asave()
person.last_name = "Roe"
await person.asave()

latest_version, old_version = await Person.objects.using(
    LAKEHOUSE_DB_ALIAS,
).order_by('-_metadata__updated_at').aexecute()


# Two versions of the same object have the same object_id
assert (await old_version.aget_metadata()).address.object_id == (await latest_version.aget_metadata()).address.object_id
# But different object_versions
assert (await old_version.get_metadata()).address.object_version != (await latest_version.aget_metadata()).address.object_version

Object history

Every time you save an object, AMSDAL creates a new version of it. Different versions of the same object have the same object_id, but different object_version. Each version has a prior_version and next_version fields that point to the previous and next versions of the same object. For the latest version, the next_version is None, and for the first version, the prior_version is None.

# The first version has no prior version, and have next version
assert old_version.get_metadata().prior_version is None
assert old_version.get_metadata().next_version == latest_version.get_metadata().address.object_version

# And vice versa for the latest version
assert latest_version.get_metadata().prior_version == old_version.get_metadata().address.object_version
assert latest_version.get_metadata().next_version is None
# The first version has no prior version, and have next version
assert (await old_version.aget_metadata()).prior_version is None
assert (await old_version.aget_metadata()).next_version == (await latest_version.aget_metadata()).address.object_version

# And vice versa for the latest version
assert (await latest_version.aget_metadata()).prior_version == (await old_version.aget_metadata()).address.object_version
assert (await latest_version.aget_metadata()).next_version is None

References

Metadata has a reference_to and referenced_by fields that store all references to the objects. It's a list of Reference objects.

from models.user.person import Person
from models.user.location import Location

# Create a person
jane = Person(first_name="Jane")
jane.save()

# Initially, person has no references
print(jane.get_metadata().referenced_by)
#> []
print(jane.get_metadata().reference_to)
#> []

# Create a location
location = Location(name="New York")
location.save()

# Location has no references as well
print(location.get_metadata().referenced_by)
#> []
print(location.get_metadata().reference_to)
#> []

# Now let's create a reference from person to location
jane.location = location
jane.save()

# Now person has a reference to the latest version of location
print(jane.get_metadata().reference_to)
#> [Reference(ref=connection#Location:LATEST:6a2600583f0d4723ae6e3feb450e7f57:LATEST)]
print(jane.get_metadata().referenced_by)
#> []

# And location has a reference from person
# Please note that the reference is the specific version of the person, that referenced the location
# Another interesting thing is that we never saved the location, but it now has a reference from person
print(location.get_metadata().referenced_by)
#> [Reference(ref=sqlite_connection#Person:0f6f012576f84d35a757cc4aac8b2fd9:b37ef7211db340eea7761388548b60f5:3a0ad633f3324636ab2661fa7d457b19)]
print(location.get_metadata().reference_to)
#> []
from models.user.person import Person
from models.user.location import Location

# Create a person
jane = Person(first_name="Jane")
await jane.asave()

# Initially, person has no references
print((await jane.aget_metadata()).referenced_by)
#> []
print((await jane.aget_metadata()).reference_to)
#> []

# Create a location
location = Location(name="New York")
await location.asave()

# Location has no references as well
print((await location.aget_metadata()).referenced_by)
#> []
print((await location.aget_metadata()).reference_to)
#> []

# Now let's create a reference from person to location
jane.location = location
await jane.asave()

# Now person has a reference to the latest version of location
print((await jane.aget_metadata()).reference_to)
#> [Reference(ref=connection#Location:LATEST:6a2600583f0d4723ae6e3feb450e7f57:LATEST)]
print((await jane.aget_metadata()).referenced_by)
#> []

# And location has a reference from person
# Please note that the reference is the specific version of the person, that referenced the location
# Another interesting thing is that we never saved the location, but it now has a reference from person
print((await location.aget_metadata()).referenced_by)
#> [Reference(ref=sqlite_connection#Person:0f6f012576f84d35a757cc4aac8b2fd9:b37ef7211db340eea7761388548b60f5:3a0ad633f3324636ab2661fa7d457b19)]
print((await location.aget_metadata()).reference_to)
#> []

Timestamps

Each record has a created_at and updated_at fields that store the timestamps of the record creation and last update. Different versions of the same object have the same created_at timestamp, but different updated_at timestamps.

from models.user.person import Person

person = Person(first_name="Jane")
person.save()

initial_created_at = person.get_metadata().created_at
initial_updated_at = person.get_metadata().updated_at

person.last_name = "Roe"
person.save()

assert person.get_metadata().created_at == initial_created_at
assert person.get_metadata().updated_at > initial_updated_at
from models.user.person import Person

person = Person(first_name="Jane")
await person.asave()

initial_created_at = (await person.aget_metadata()).created_at
initial_updated_at = (await person.aget_metadata()).updated_at

person.last_name = "Roe"
await person.asave()

assert (await person.aget_metadata()).created_at == initial_created_at
assert (await person.aget_metadata()).updated_at > initial_updated_at