Skip to content

Model classes

API Documentation

amsdal_models.classes.model.Model

AMSDAL models inherit from the AMSDAL Model class, that itself inherits from Pydantic's BaseModel providing uniform functionality to:

  • validate data against model schemas
  • create and store new records
  • query data
  • edit, delete existing records
  • serialize any record with type hints for fast and reliable coding and integration with your favorite IDE

AMSDAL automtically builds Python models from defined JSON schemas.

Creating objects

To create a new instance of a model, instantiate it like any other Python class:

from models.user.person import Person

person = Person(first_name="John", last_name="Doe")

When an object is instantiated its validations will be applied (e.g. by options or required model schema configuration) but the object will not be saved to the connection and unique validations will not be applied TBD.

To save a record to the connection you simply call the save() method:

from models.user.person import Person

person = Person(first_name="John", last_name="Doe")
person.save()
from models.user.person import Person

person = Person(first_name="John", last_name="Doe")
await person.asave()

Allowing you to initialize and apply transformation logic before saving the record to the connection (i.e. ETL):

def create_person(first_name: str, last_name: str, country_code: str):
    from models.user.person import Person

    person = Person(first_name=first_name, last_name=last_name)

    if country_code == "US":
        person.country = "United States"
    elif country_code == "GB":
        person.country = "United Kingdom"
    else:
        person.country = "Unknown"

    person.save()

    return person
async def create_person(first_name: str, last_name: str, country_code: str):
    from models.user.person import Person

    person = Person(first_name=first_name, last_name=last_name)

    if country_code == "US":
        person.country = "United States"
    elif country_code == "GB":
        person.country = "United Kingdom"
    else:
        person.country = "Unknown"

    await person.asave()

    return person

Warning

If an object is not saved as a record it cannot be used as a reference by another other object.

Updating objects

To update an existing object, you can also use the save() method:

from models.user.person import Person

# retrieve the object from DB
person = Person.objects.get(first_name="John", last_name="Doe").execute()

# change the properties
person.country = "United States"

# save the changes to the DB
person.save()
from models.user.person import Person

# retrieve the object from DB
person = await Person.objects.get(first_name="John", last_name="Doe").aexecute()

# change the properties
person.country = "United States"

# save the changes to the DB
await person.asave()

Each time you call the save() method, a new version of the object will be saved to the connectiion, either overwriting the existing record in the case of a state connection or as a new version in the case of a historical connection.

Warning

If you do not call the save() method, the changes will not be saved to the connection.

Note

Regardless of whether you are using a state connection, a history of your data will automatically be stored in the AMSDAL Lakehouse

Deleting objects

To delete an existing object, use the delete() method:

from models.user.person import Person

# retrieve the object from DB
person = Person.objects.get(first_name="John", last_name="Doe").execute()

# delete the object from the DB
person.delete()
from models.user.person import Person

# retrieve the object from DB
person = await Person.objects.get(first_name="John", last_name="Doe").aexecute()

# delete the object from the DB
await person.adelete()

Where in the case of a state connection the record will be deleted and in the case of a historical connection the new version of this object will be created with the deleted flag set to True.

Frozen objects

By marking a model as frozen, in the case of a historical connection, AMSDAL allows for a property reference to be set to a specific version of an object

from models.user.person import Person

person = Person.objects.get(first_name="John", last_name="Doe").execute()
person.frozen()  # mark the model as frozen
from models.user.person import Person

person = await Person.objects.get(first_name="John", last_name="Doe").aexecute()
person.frozen()  # mark the model as frozen

AMSDAL will throw an error if you change the properties of the model or save

Note

You cannot mark the object as frozen if it is not saved to the DB yet.

How it works:

When you mark the model as frozen, it has not impact on the data in the connection, it simply marks the object as frozen in the memory so that when you use this model as the value of a reference in another other objects this reference will freeze the specific version id.

Frozen objects are useful when you want to lock a reference by another objects, so that if an object is updated in the future, the reference will still point to a static version of the object.

For example:

from models.user.person import Person
from models.user.event import Event

person = Person.objects.get(first_name="John", last_name="Doe", age=18).execute()
person.frozen()

event = Event(name="Birthday 18", person=person, date="2023-02-20")
event.save()
from models.user.person import Person
from models.user.event import Event

person = await Person.objects.get(first_name="John", last_name="Doe", age=18).aexecute()
person.frozen()

event = Event(name="Birthday 18", person=person, date="2023-02-20")
await event.asave()

If in the future the age of this person is changed, this event record will still point to the version of the person with an age of 18.

References

When accessing an attribute of a model that is a reference to another model, AMSDAL will automatically retrieve the referenced object from the connection.

from models.user.person import Person
from models.user.event import Event

event = Event.objects.get(name="Birthday 18").execute()
print(event.person.first_name)
#> John
from models.user.person import Person
from models.user.event import Event

event = await Event.objects.get(name="Birthday 18").aexecute()
print(event.person.first_name)
#> John

You can access the reference of a model with metadata. When dealing with references, you can still get target object with ReferenceLoader.

from amsdal_models.classes.helpers.reference_loader import ReferenceLoader

person_reference = event.person.get_metadata().reference
person = ReferenceLoader(person_reference).load_reference()
print(person.first_name)
#> John
from amsdal_models.classes.helpers.reference_loader import ReferenceLoader

person_reference = event.person.get_metadata().reference
person = await ReferenceLoader(person_reference).aload_reference()
print(person.first_name)
#> John

Alternatively you can just use __await__ method of the Reference object.

person_reference = event.person.get_metadata().reference
person = await person_reference
print(person.first_name)
#> John

Meta-Class

Note

Documentation coming soon!