Working with Models

Once you've defined a model, you can create, read, update, and delete records. All models inherit from Model, which provides the objects manager for querying, instance methods for persistence, and built-in version history.

All operations have both sync and async variants.

---

Creating Objects

Instantiate a model and call save() to persist it:

SyncAsync

person = Person(first_name='John', last_name='Doe')
person.country = 'United States'
person.save()

person = Person(first_name='John', last_name='Doe')
person.country = 'United States'
await person.asave()

Field validation (types, required fields, validators) runs at instantiation. The record is written to the database only when you call save() / asave().

Nested references are saved automatically — if a referenced object hasn't been saved yet, save() persists it first:

event = Event(name='Birthday', person=Person(first_name='John', last_name='Doe'))
event.save()  # saves Person first, then Event

Warning

Unsaved objects cannot be used in queries. Person.objects.filter(profile=unsaved_profile) won't match anything because the object doesn't exist in the database yet.

Bulk Create

For creating multiple records efficiently:

SyncAsync

people = [
    Person(first_name='Alice', last_name='Smith'),
    Person(first_name='Bob', last_name='Jones'),
]
Person.objects.bulk_create(people)

await Person.objects.bulk_acreate(people)

---

Updating Objects

Fetch an object, modify it, and call save():

SyncAsync

person = Person.objects.get(first_name='John', last_name='Doe').execute()
person.country = 'United Kingdom'
person.save()

person = await Person.objects.get(first_name='John', last_name='Doe').aexecute()
person.country = 'United Kingdom'
await person.asave()

Each save() creates a new version of the object. With a historical connection, all previous versions are preserved. With a state connection, the record is overwritten but history is still tracked in the Lakehouse.

Bulk Update

SyncAsync

for person in people:
    person.country = 'Canada'
Person.objects.bulk_update(people)

await Person.objects.bulk_aupdate(people)

Refetch from Database

Reload the object's current state from the database:

SyncAsync

person = person.refetch_from_db()

# Or fetch the latest version specifically:
person = person.refetch_from_db(latest=True)

person = await person.arefetch_from_db()

# Or fetch the latest version specifically:
person = await person.arefetch_from_db(latest=True)

---

Deleting Objects

SyncAsync

person = Person.objects.get(first_name='John', last_name='Doe').execute()
person.delete()

person = await Person.objects.get(first_name='John', last_name='Doe').aexecute()
await person.adelete()

With a state connection, the record is removed. With a historical connection, a new version is created with a deleted flag.

Bulk Delete

SyncAsync

Person.objects.bulk_delete(people)

await Person.objects.bulk_adelete(people)

---

Version History

AMSDAL tracks every change as a version. You can navigate between versions of any object:

SyncAsync

person = Person.objects.get(first_name='John').execute()

# Get the previous version
prev = person.previous_version()

# Get the next version (if navigating from an older version)
nxt = person.next_version()

person = await Person.objects.get(first_name='John').aexecute()
prev = await person.aprevious_version()
nxt = await person.anext_version()

Both return None if there is no previous/next version.

Fetching a Specific Version

SyncAsync

specific = Person.objects.get_specific_version(
    object_id='abc123',
    object_version='v2',
)

specific = await Person.objects.aget_specific_version(
    object_id='abc123',
    object_version='v2',
)

---

Relationships

For the full reference on relation accessors, RelatedSet semantics, and lakehouse historical reads, see Relationships.

Forward References (FK)

When a model has a forward reference (foreign key) to another model, AMSDAL loads the referenced object automatically on access:

SyncAsync

class Event(Model):
    name: str
    person: Person

event = Event.objects.get(name='Birthday').execute()
print(event.person.first_name)
#> John

event = await Event.objects.get(name='Birthday').aexecute()
person = await event.person  # awaits the forward-FK resolver
print(person.first_name)
#> John

Note

In async mode, forward-FK access (event.person) returns an awaitable because the reference is loaded via ReferenceLoader.aload_reference(). Once awaited, the resolved instance is cached on the parent — subsequent access of the same FK on the same instance returns the cached value directly (still as an awaitable for API consistency).

Frozen References

By default, a reference points to the latest version of the referenced object. To pin a reference to a specific version, pass a frozen Reference object instead of the model instance:

SyncAsync

person = Person.objects.get(first_name='John', age=18).execute()

# Create a frozen reference to the current version
frozen_ref = person.build_reference(is_frozen=True)

event = Event(name='Birthday 18', person=frozen_ref, date='2023-02-20')
event.save()

person = await Person.objects.get(first_name='John', age=18).aexecute()

frozen_ref = await person.abuild_reference(is_frozen=True)

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

If person.age is updated later, this event still references the version where age=18.

ReferenceLoader

You can load objects from a Reference object using ReferenceLoader:

SyncAsync

from amsdal_models.classes.helpers.reference_loader import ReferenceLoader

ref = person.build_reference()
loaded_person = ReferenceLoader(ref).load_reference()

from amsdal_models.classes.helpers.reference_loader import ReferenceLoader

ref = await person.abuild_reference()
loaded_person = await ReferenceLoader(ref).aload_reference()

Reverse References

Whenever a model declares a forward FK to another model, the target gains a reverse accessor automatically. Given:

class Author(Model):
    name: str

class Book(Model):
    title: str
    author: Author

Author automatically gets a book_set attribute:

SyncAsync

author = Author.objects.get(name='Ann').execute()
books = list(author.book_set)            # iterate
count = author.book_set.count()          # SELECT COUNT(*)
has_any = author.book_set.exists()       # SELECT 1 LIMIT 1
first = author.book_set[0]               # SELECT ... LIMIT 1 OFFSET 0

author = await Author.objects.get(name='Ann').aexecute()
books = [b async for b in author.book_set]   # iterate (awaits internally)
count = await author.book_set.acount()
has_any = await author.book_set.aexists()

Naming: the default reverse-name is <reverser_lower>_set (Author.book_set). Override or disable per field:

class Book(Model):
    author: Author = ReferenceField(related_name='books')   # → author.books

class Note(Model):
    author: Author = ReferenceField(related_name='+')       # disabled

Two FKs that resolve to the same reverse-name on a target raise AmsdalReverseFKConflictError at class-build time — add an explicit related_name to one of them.

The registry is exposed as Author.__reverse_foreign_keys__: dict[str, ReverseFKDescriptor].

Warning

Passing a reverse-FK keyword to a model constructor (e.g. Author(book_set=[...])) raises AmsdalReverseFKConstructorError. Use .add(...) / .set(...) on the accessor instead, then save() the parent to flush.

RelatedSet (M2M and reverse-FK accessors)

Both reverse-FK (author.book_set) and many-to-many (post.tags) accessors return a RelatedSet[T] — a list subclass with cache-aware deferred semantics.

When the cache is populated (via prefetch or add()), reads use it directly; otherwise reads issue optimal SQL:

Operation	SQL when cache empty
len(rs) / rs.count()	SELECT COUNT(*)
bool(rs) / rs.exists()	SELECT 1 ... LIMIT 1
rs[i] / rs[i:j]	SELECT ... LIMIT/OFFSET
for x in rs / list(rs)	SELECT * (full materialization)

Writes (add / remove / set / clear) are synchronous regardless of mode and mutate an in-memory unit-of-work buffer. The DB is touched only when parent.save() / await parent.asave() flushes the deltas, inside the same transaction as the parent save:

post = Post.objects.get(title='Hello').execute()
post.tags.add(t1, t2)        # in-memory only
post.tags.remove(old_tag)    # in-memory only
post.save()                  # flushes m2m through-rows atomically with the post

Async note: write methods (add / remove / set / clear) are sync in any mode — there are no aadd / aremove / aset / aclear siblings. Reads have async siblings for the named methods: use acount() and aexists() in async mode. Iteration, indexing, in, bool, len work in any mode when the cache is populated; in async mode without cache they raise — use await rs (or async for x in rs) to populate first.

Non-nullable reverse-FK detach behavior: remove() / clear() / shrinking set() on a non-nullable reverse-FK silently buffer the change; at parent.save() the flush tries to set the child's FK column to None, which fails Pydantic validation and raises pydantic.ValidationError (with an error entry pointing to the FK field, input is None).

To actually detach a child from a non-nullable parent, either re-parent it to a different valid parent, or delete the child outright:

import pytest
from pydantic import ValidationError

# ❌ Detaching a child whose FK is non-nullable — fails at flush
c.employee_set.remove(e)
with pytest.raises(ValidationError):
    c.save()

# ✅ Option 1 — reparent the child to a different valid company
c2.employee_set.add(e)
c2.save()         # employee.company is now c2 — no nullification

# ✅ Option 2 — hard-delete the child
e.delete()

See QuerySet for the full filter API returned by rs.filter(...) / rs.exclude(...) / rs.order_by(...).

---

Serialization

Models provide Pydantic's model_dump() and additional AMSDAL-specific methods for serializing with references:

# Standard Pydantic serialization
data = person.model_dump()
json_str = person.model_dump_json()

# With references (foreign keys serialized as reference objects)
data = person.model_dump_refs()
json_str = person.model_dump_json_refs()

Note

When async_mode=True, model_dump() and model_dump_json() automatically serialize references as $ref dictionaries (same behavior as model_dump_refs()), because synchronous reference loading is not allowed in async mode. To inline referenced objects in async mode, you must load them explicitly first (await event.person) and re-serialize, or use the sync code path.

---

ExternalModel

ExternalModel provides read-only access to existing database tables that AMSDAL doesn't manage. No migrations, no versioning — just querying:

from amsdal_models.classes.external_model import ExternalModel

class LegacyUser(ExternalModel):
    __table_name__ = 'users'
    __connection__ = 'legacy_db'

    id: int
    username: str
    email: str

By default, ExternalModel assumes a single-column PK named id. For tables with a different PK column or a composite PK, set __primary_key__:

class Order(ExternalModel):
    __table_name__ = 'orders'
    __connection__ = 'legacy_db'
    __primary_key__ = ['order_id']           # single, custom name

    order_id: int
    customer: str

class OrderLine(ExternalModel):
    __table_name__ = 'order_lines'
    __connection__ = 'legacy_db'
    __primary_key__ = ['order_id', 'line_no']  # composite

    order_id: int
    line_no: int
    sku: str

Query it like a regular model:

SyncAsync

users = LegacyUser.objects.filter(username='alice').execute()

users = await LegacyUser.objects.filter(username='alice').aexecute()

ExternalModel does not have save(), delete(), lifecycle hooks, or metadata. It's designed for integrating with external databases you don't control.

Feature	Model	ExternalModel
CRUD operations	Full	Read-only
Version history	Yes	No
Migrations	Yes	No
Lifecycle hooks	Yes	No
References	Yes	No
objects manager	Full Manager	ExternalManager