Model Definition

Define your data models as Python classes by inheriting from Model. Each model maps to a database table — AMSDAL handles schema generation, migrations, validation, and querying automatically.

from amsdal.models import Model


class Person(Model):
    first_name: str
    last_name: str

This generates:

CREATE TABLE Person (
    partition_key TEXT NOT NULL,
    first_name TEXT,
    last_name TEXT,
    PRIMARY KEY (partition_key)
)

Models are built on top of Pydantic, so you get runtime type validation, serialization, and all Pydantic features out of the box.

You can import Model and related classes from amsdal.models.

---

Defining Fields

Required Fields

By default, fields without a default value are required:

from amsdal.models import Model

class Person(Model):
    first_name: str
    last_name: str

Optional Fields

Use Optional or the | None syntax to allow None values:

Using typingUsing native Python

from typing import Optional
from amsdal.models import Model

class Person(Model):
    first_name: Optional[str]
    last_name: Optional[str]

from amsdal.models import Model

class Person(Model):
    first_name: str | None
    last_name: str | None

Default Values

Assign defaults directly or use Pydantic's Field:

Direct assignmentPydantic Field

from amsdal.models import Model

class Person(Model):
    first_name: str = 'John'
    last_name: str = 'Doe'

from pydantic import Field
from amsdal.models import Model

class Person(Model):
    first_name: str = Field(default='John')
    last_name: str = Field(default='Doe')

---

Relationships

Many-to-One

Reference another model directly as a field type:

from amsdal.models import Model

class Asset(Model):
    asset_name: str

class Person(Model):
    first_name: str
    last_name: str
    asset: Asset

This creates a foreign key — by default the database column is named asset_partition_key.

To customize the foreign key column name, use ReferenceField:

from amsdal.models import Model, ReferenceField

class Person(Model):
    first_name: str
    last_name: str
    asset: Asset = ReferenceField(..., db_field='asset_id')

ReferenceField extends Pydantic's Field, so it accepts all the same arguments.

Composite Foreign Keys

When the referenced model has a composite primary key:

from amsdal.models import Model, ReferenceField

class Asset(Model):
    __primary_key__ = ['asset_name', 'asset_type']

    asset_name: str
    asset_type: str

class Person(Model):
    first_name: str
    last_name: str
    asset: Asset = ReferenceField(..., db_field=['asset_name', 'asset_type'])

Generated schema:

CREATE TABLE Person (
    partition_key TEXT NOT NULL,
    first_name TEXT,
    last_name TEXT,
    asset_name TEXT,
    asset_type TEXT,
    PRIMARY KEY (partition_key),
    FOREIGN KEY (asset_name, asset_type) REFERENCES Asset(asset_name, asset_type)
)

On Delete Behavior

Control what happens when a referenced object is deleted using the on_delete parameter:

from amsdal.models import Model, ReferenceField
from amsdal_models.classes.relationships.enum import ReferenceMode

class Person(Model):
    first_name: str
    last_name: str
    asset: Asset = ReferenceField(..., on_delete=ReferenceMode.CASCADE)

Available modes:

Mode	Behavior
CASCADE	Delete this object too
PROTECT	Prevent deletion of the referenced object
RESTRICT	Similar to PROTECT, checked at database level
SET_NULL	Set the foreign key to NULL
SET_DEFAULT	Set the foreign key to its default value
DO_NOTHING	Do nothing (may cause integrity errors)

Many-to-Many

Use list[Model] to define a many-to-many relationship:

from amsdal.models import Model

class Asset(Model):
    name: str

class Person(Model):
    first_name: str
    last_name: str
    assets: list[Asset]

AMSDAL auto-generates a junction table:

CREATE TABLE Asset (
    partition_key TEXT NOT NULL,
    name TEXT,
    PRIMARY KEY (partition_key)
)

CREATE TABLE Person (
    partition_key TEXT NOT NULL,
    first_name TEXT,
    last_name TEXT,
    PRIMARY KEY (partition_key)
)

CREATE TABLE PersonAsset (
    person_partition_key TEXT NOT NULL,
    asset_partition_key TEXT NOT NULL,
    PRIMARY KEY (person_partition_key, asset_partition_key),
    FOREIGN KEY (person_partition_key) REFERENCES Person(partition_key),
    FOREIGN KEY (asset_partition_key) REFERENCES Asset(partition_key)
)

Custom Junction Model

To add extra fields to the junction table, define the through model explicitly using ManyReferenceField:

from amsdal.models import Model, ManyReferenceField, ReferenceField

class Asset(Model):
    name: str

class PersonAsset(Model):
    person: 'Person' = ReferenceField(db_field='person_id')
    asset: Asset = ReferenceField(db_field='asset_id')
    is_favorite: bool = False

class Person(Model):
    first_name: str
    last_name: str
    assets: list[Asset] = ManyReferenceField(
        through=PersonAsset,
        through_fields=('person', 'asset'),
    )

• through — the junction model class
• through_fields — a tuple of field names in the junction model: (field pointing to this model, field pointing to the related model)

---

Table Configuration

Control how your model maps to the database using class-level attributes.

Custom Table Name

from typing import ClassVar
from amsdal.models import Model

class Person(Model):
    __table_name__: ClassVar[str] = 'people'

    first_name: str
    last_name: str

Generates CREATE TABLE people (...) instead of CREATE TABLE Person (...).

Custom Primary Key

By default, AMSDAL adds a partition_key field as the primary key. Override it with __primary_key__:

from typing import ClassVar
from amsdal.models import Model

class Person(Model):
    __primary_key__: ClassVar[list[str]] = ['person_id']

    person_id: int
    first_name: str
    last_name: str

Composite primary key:

from typing import ClassVar
from amsdal.models import Model

class Person(Model):
    __primary_key__: ClassVar[list[str]] = ['first_name', 'last_name']

    first_name: str
    last_name: str

Indexes

from typing import ClassVar
from amsdal.models import Model, IndexInfo

class Person(Model):
    __indexes__: ClassVar[list[IndexInfo]] = [
        IndexInfo(field='first_name', name='idx_first_name'),
    ]

    first_name: str
    last_name: str

Unique Constraints

from typing import ClassVar
from amsdal.models import Model, UniqueConstraint

class Person(Model):
    __constraints__: ClassVar[list[UniqueConstraint]] = [
        UniqueConstraint(fields=['first_name', 'last_name'], name='unique_full_name'),
    ]

    first_name: str
    last_name: str

---

Inheritance

Models support standard Python class inheritance. The child model gets all parent fields plus its own, with a foreign key back to the parent table:

from amsdal.models import Model

class Person(Model):
    first_name: str
    last_name: str

class Employee(Person):
    company_name: str

Generated schema:

CREATE TABLE Person (
    partition_key TEXT NOT NULL,
    first_name TEXT,
    last_name TEXT,
    PRIMARY KEY (partition_key)
)

CREATE TABLE Employee (
    partition_key TEXT NOT NULL,
    first_name TEXT,
    last_name TEXT,
    company_name TEXT,
    PRIMARY KEY (partition_key),
    FOREIGN KEY (partition_key) REFERENCES Person(partition_key)
)

---

Validators

Options Validator

Restrict a field to a predefined set of values:

from typing import Any
from pydantic.functional_validators import field_validator

from amsdal.models import Model
from amsdal.models import validate_options

class Person(Model):
    first_name: str
    last_name: str
    gender: str

    @field_validator('gender')
    @classmethod
    def validate_gender(cls: type, value: Any) -> Any:
        return validate_options(
            value,
            options=['Male', 'Female'],
        )

Using validate_options instead of a plain Literal type allows AMSDAL to track and store these options in the lakehouse database.

Non-Empty Key Dictionary Validator

Ensure all keys in a dictionary field are non-empty strings:

from typing import Any
from pydantic.functional_validators import field_validator

from amsdal.models import Model
from amsdal.models import validate_non_empty_keys

class Person(Model):
    first_name: str
    last_name: str
    equipment: dict[str, str]

    @field_validator('equipment')
    @classmethod
    def validate_equipment_keys(cls: type, value: Any) -> Any:
        return validate_non_empty_keys(value)