Model classes¶
API Documentation
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!