Models API

Model

Model

Bases: BaseModel

Base model class for RestMachine ORM.

Provides ActiveRecord-style CRUD operations and integrates with Pydantic for validation. Models are compatible with RestMachine for automatic API integration.

Example

class User(Model): ... model_backend: ClassVar[Backend] = InMemoryBackend() ... ... id: str = Field(primary_key=True) ... email: str = Field(unique=True) ... name: str ... user = User.create(id="123", email="alice@example.com", name="Alice") user.name = "Alice Smith" user.save()

Alternative syntax using class parameter:

class User(Model, model_backend=InMemoryBackend()): ... id: str = Field(primary_key=True) ... name: str

Functions

__init_subclass__

__init_subclass__(model_backend: Optional[Backend] = None, **kwargs: Any)

Collect hooks from mixins when model class is defined.

Parameters:

Name	Type	Description	Default
model_backend	Optional[Backend]	Optional backend to use for this model (alternative to ClassVar)	None
**kwargs	Any	Additional arguments passed to parent	{}

Source code in packages/restmachine-orm/src/restmachine_orm/models/base.py

def __init_subclass__(cls, model_backend: Optional["Backend"] = None, **kwargs: Any):
    """
    Collect hooks from mixins when model class is defined.

    Args:
        model_backend: Optional backend to use for this model (alternative to ClassVar)
        **kwargs: Additional arguments passed to parent
    """
    super().__init_subclass__(**kwargs)

    # Support class parameter pattern: class User(Model, model_backend=InMemoryBackend())
    if model_backend is not None:
        cls.model_backend = model_backend

    # Reset class variables for this specific subclass
    cls._before_save_hooks = []
    cls._after_save_hooks = []
    cls._after_load_hooks = []
    cls._query_methods = {}
    cls._query_operators = {}
    cls._auto_query_filters = []
    cls._geo_field_names = []

    # Collect hooks from all bases (mixins) and the current class
    for base in [cls] + list(cls.__bases__):
        # Collect hooks from mixin methods
        for attr_name in dir(base):
            # Skip special Python attributes (double underscore)
            if attr_name.startswith('__'):
                continue
            try:
                attr = getattr(base, attr_name)
                if callable(attr) and getattr(attr, '_is_before_save_hook', False):
                    if attr not in cls._before_save_hooks:
                        cls._before_save_hooks.append(attr)
                elif callable(attr) and getattr(attr, '_is_after_save_hook', False):
                    if attr not in cls._after_save_hooks:
                        cls._after_save_hooks.append(attr)
                elif callable(attr) and getattr(attr, '_is_after_load_hook', False):
                    if attr not in cls._after_load_hooks:
                        cls._after_load_hooks.append(attr)
                elif callable(attr) and getattr(attr, '_is_query_method', False):
                    method_name = getattr(attr, '_query_method_name', attr.__name__)
                    cls._query_methods[method_name] = attr
                elif callable(attr) and getattr(attr, '_is_query_operator', False):
                    # Type-based operator
                    if hasattr(attr, '_operator_type'):
                        op_type = getattr(attr, '_operator_type')
                        op_name = getattr(attr, '_operator_name')
                        # We'll map this to field names after fields are set
                        if not hasattr(cls, '_type_operators'):
                            cls._type_operators = {}  # type: ignore[attr-defined]
                        cls._type_operators[(op_type, op_name)] = attr  # type: ignore[attr-defined]
                    elif hasattr(attr, '_operator_types'):
                        for op_type in getattr(attr, '_operator_types'):
                            op_name = getattr(attr, '_operator_name')
                            if not hasattr(cls, '_type_operators'):
                                cls._type_operators = {}  # type: ignore[attr-defined]
                            cls._type_operators[(op_type, op_name)] = attr  # type: ignore[attr-defined]
            except AttributeError:
                continue

        # Collect auto query filters (from ExpirationMixin, etc.)
        if hasattr(base, '_auto_query_filters'):
            cls._auto_query_filters.extend(base._auto_query_filters)

    # Map type-based operators to field names
    if hasattr(cls, '_type_operators') and hasattr(cls, 'model_fields'):
        cls._map_type_operators_to_fields()

create classmethod

create(**kwargs: Any) -> Model

Create and save a new record in one operation.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Field values for the new record	{}

Returns:

Type	Description
Model	Created and saved model instance

Raises:

Type	Description
ValidationError	If field validation fails
DuplicateKeyError	If unique constraint is violated

Example

user = User.create(id="123", email="alice@example.com", name="Alice")

Source code in packages/restmachine-orm/src/restmachine_orm/models/base.py

@classmethod
def create(cls, **kwargs: Any) -> "Model":
    """
    Create and save a new record in one operation.

    Args:
        **kwargs: Field values for the new record

    Returns:
        Created and saved model instance

    Raises:
        ValidationError: If field validation fails
        DuplicateKeyError: If unique constraint is violated

    Example:
        >>> user = User.create(id="123", email="alice@example.com", name="Alice")
    """
    instance = cls(**kwargs)
    instance.save()
    return instance

upsert classmethod

upsert(**kwargs: Any) -> Model

Create or update a record (upsert).

If a record with the same key exists, it will be overwritten. Unlike create(), this does not raise DuplicateKeyError.

Callbacks

• Calls all @before_save methods before persisting
• Calls all @after_save methods after persisting

Parameters:

Name	Type	Description	Default
**kwargs	Any	Field values for the record	{}

Returns:

Type	Description
Model	Upserted and saved model instance

Raises:

Type	Description
ValidationError	If field validation fails

Example

user = User.upsert(id="123", email="alice@example.com", name="Alice")

If user with id=123 exists, it will be overwritten

Source code in packages/restmachine-orm/src/restmachine_orm/models/base.py

@classmethod
def upsert(cls, **kwargs: Any) -> "Model":
    """
    Create or update a record (upsert).

    If a record with the same key exists, it will be overwritten.
    Unlike create(), this does not raise DuplicateKeyError.

    Callbacks:
        - Calls all @before_save methods before persisting
        - Calls all @after_save methods after persisting

    Args:
        **kwargs: Field values for the record

    Returns:
        Upserted and saved model instance

    Raises:
        ValidationError: If field validation fails

    Example:
        >>> user = User.upsert(id="123", email="alice@example.com", name="Alice")
        >>> # If user with id=123 exists, it will be overwritten
    """
    instance = cls(**kwargs)

    # Call hook-based before_save methods (from mixins)
    for hook in cls._before_save_hooks:
        hook(instance)

    # Call all registered before_save callbacks
    for callback in cls._before_save_callbacks:
        callback(instance)

    # Validate the model before saving
    instance.model_validate(instance.model_dump())

    backend = cls._get_backend()
    data = instance.model_dump()

    # Backend returns the data that was stored
    result_data = backend.upsert(cls, data)

    # Update instance with any fields set by backend extensions (e.g., timestamps)
    for key, value in result_data.items():
        if hasattr(instance, key):
            setattr(instance, key, value)

    # Mark instance as persisted
    instance._is_persisted = True

    # Call hook-based after_save methods (from mixins)
    for hook in cls._after_save_hooks:
        hook(instance)

    # Call all registered after_save callbacks
    for callback in cls._after_save_callbacks:
        callback(instance)

    return instance

save

save() -> Model

Validate and save this record to the database.

Creates a new record if not persisted, otherwise updates existing.

Callbacks

• Calls all @before_save methods before persisting
• Calls all @after_save methods after persisting

Returns:

Type	Description
Model	Self for method chaining

Raises:

Type	Description
ValidationError	If field validation fails

Example

user = User(id="123", name="Alice") user.save() # Validates and saves

user.name = "Alice Smith" user.save() # Validates and updates

Source code in packages/restmachine-orm/src/restmachine_orm/models/base.py

def save(self) -> "Model":
    """
    Validate and save this record to the database.

    Creates a new record if not persisted, otherwise updates existing.

    Callbacks:
        - Calls all @before_save methods before persisting
        - Calls all @after_save methods after persisting

    Returns:
        Self for method chaining

    Raises:
        ValidationError: If field validation fails

    Example:
        >>> user = User(id="123", name="Alice")
        >>> user.save()  # Validates and saves

        >>> user.name = "Alice Smith"
        >>> user.save()  # Validates and updates
    """
    # Call hook-based before_save methods (from mixins)
    for hook in self.__class__._before_save_hooks:
        hook(self)

    # Call all registered before_save callbacks
    for callback in self.__class__._before_save_callbacks:
        callback(self)

    # Validate the model before saving
    # Pydantic automatically validates on construction and assignment
    # but we trigger it explicitly here to ensure consistency
    self.model_validate(self.model_dump())

    backend = self._get_backend()
    data = self.model_dump()

    if not self._is_persisted:
        # Create new record
        result_data = backend.create(self.__class__, data)
        self._is_persisted = True
        # Update instance with any fields set by backend extensions (e.g., timestamps)
        for key, value in result_data.items():
            if hasattr(self, key):
                setattr(self, key, value)
    else:
        # Update existing record
        result_data = backend.update(self.__class__, self)
        # Update instance with any fields modified by backend extensions
        for key, value in result_data.items():
            if hasattr(self, key):
                setattr(self, key, value)

    # Call hook-based after_save methods (from mixins)
    for hook in self.__class__._after_save_hooks:
        hook(self)

    # Call all registered after_save callbacks
    for callback in self.__class__._after_save_callbacks:
        callback(self)

    return self

delete

delete() -> bool

Delete this record from the database.

Returns:

Type	Description
bool	True if deleted successfully

Example

user.delete()

Source code in packages/restmachine-orm/src/restmachine_orm/models/base.py

def delete(self) -> bool:
    """
    Delete this record from the database.

    Returns:
        True if deleted successfully

    Example:
        >>> user.delete()
    """
    backend = self._get_backend()
    # Backend handles key extraction from the model instance
    return backend.delete(self.__class__, self)

get classmethod

get(**filters: Any) -> Optional[Model]

Get a single record by primary key or filters.

Parameters:

Name	Type	Description	Default
**filters	Any	Filter conditions (typically primary key)	{}

Returns:

Type	Description
Optional[Model]	Model instance, or None if not found

Example

user = User.get(id="123") if user: ... print(user.name)

Source code in packages/restmachine-orm/src/restmachine_orm/models/base.py

@classmethod
def get(cls, **filters: Any) -> Optional["Model"]:
    """
    Get a single record by primary key or filters.

    Args:
        **filters: Filter conditions (typically primary key)

    Returns:
        Model instance, or None if not found

    Example:
        >>> user = User.get(id="123")
        >>> if user:
        ...     print(user.name)
    """
    backend = cls._get_backend()
    data = backend.get(cls, **filters)
    if data:
        instance = cls(**data)
        instance._is_persisted = True

        # Call after_load hooks (for geo deserialization, etc.)
        for hook in cls._after_load_hooks:
            hook(instance)

        return instance
    return None

find_by classmethod

find_by(**conditions: Any) -> Optional[Model]

Find the first record matching conditions (eager execution).

This method executes immediately and returns the first matching record. For lazy query building, use where() instead.

Parameters:

Name	Type	Description	Default
**conditions	Any	Filter conditions (all ANDed together)	{}

Returns:

Type	Description
Optional[Model]	First matching model instance, or None if not found

Examples:

>>> # Find first user with email
>>> user = User.find_by(email="alice@example.com")
>>> if user:
...     print(user.name)

>>> # Find first active user
>>> user = User.find_by(status="active")

Source code in packages/restmachine-orm/src/restmachine_orm/models/base.py

@classmethod
def find_by(cls, **conditions: Any) -> Optional["Model"]:
    """
    Find the first record matching conditions (eager execution).

    This method executes immediately and returns the first matching record.
    For lazy query building, use where() instead.

    Args:
        **conditions: Filter conditions (all ANDed together)

    Returns:
        First matching model instance, or None if not found

    Examples:
        >>> # Find first user with email
        >>> user = User.find_by(email="alice@example.com")
        >>> if user:
        ...     print(user.name)

        >>> # Find first active user
        >>> user = User.find_by(status="active")
    """
    return cls.where(**conditions).first()

where classmethod

where(*expressions: Any, **conditions: Any) -> QueryBuilder

Create a lazy query builder for finding records.

Returns a QueryBuilder that doesn't execute until results are accessed (via .all(), .first(), .last(), iteration, etc.)

Supports both field expressions and keyword arguments.

Parameters:

Name	Type	Description	Default
*expressions	Any	Field expressions like User.age > 25	()
**conditions	Any	Keyword filter conditions (all ANDed together)	{}

Returns:

Type	Description
QueryBuilder	QueryBuilder instance for chaining

Examples:

>>> # Field expressions (new style)
>>> users = User.where(User.age > 25).all()
>>> users = User.where((User.age >= 18) & (User.age <= 65)).all()
>>> users = User.where((User.role == "admin") | (User.role == "moderator")).all()

>>> # Keyword conditions (classic style)
>>> users = User.where(status="active", age__gte=18).all()

>>> # Mixed
>>> users = User.where(User.age > 25, is_active=True).all()

>>> # Chain conditions
>>> users = User.where(age__gte=18).and_(status="active").limit(10).all()

>>> # Iterate over results (lazy)
>>> for user in User.where(User.age >= 18):
...     print(user.name)

Source code in packages/restmachine-orm/src/restmachine_orm/models/base.py

@classmethod
def where(cls, *expressions: Any, **conditions: Any) -> "QueryBuilder":
    """
    Create a lazy query builder for finding records.

    Returns a QueryBuilder that doesn't execute until results are accessed
    (via .all(), .first(), .last(), iteration, etc.)

    Supports both field expressions and keyword arguments.

    Args:
        *expressions: Field expressions like User.age > 25
        **conditions: Keyword filter conditions (all ANDed together)

    Returns:
        QueryBuilder instance for chaining

    Examples:
        >>> # Field expressions (new style)
        >>> users = User.where(User.age > 25).all()
        >>> users = User.where((User.age >= 18) & (User.age <= 65)).all()
        >>> users = User.where((User.role == "admin") | (User.role == "moderator")).all()

        >>> # Keyword conditions (classic style)
        >>> users = User.where(status="active", age__gte=18).all()

        >>> # Mixed
        >>> users = User.where(User.age > 25, is_active=True).all()

        >>> # Chain conditions
        >>> users = User.where(age__gte=18).and_(status="active").limit(10).all()

        >>> # Iterate over results (lazy)
        >>> for user in User.where(User.age >= 18):
        ...     print(user.name)
    """
    # Lazy initialization of query operators (needs to happen after Pydantic sets up model_fields)
    if hasattr(cls, '_type_operators') and not cls._query_operators:
        cls._map_type_operators_to_fields()

    backend = cls._get_backend()
    query = backend.query(cls)

    # Apply auto query filters (from ExpirationMixin, etc.)
    for auto_filter in cls._auto_query_filters:
        query = auto_filter(query)

    # Apply expressions
    if expressions:
        query = query.where(*expressions)

    # Apply keyword conditions
    if conditions:
        query = query.and_(**conditions)

    return query

all classmethod

all() -> list[Model]

Get all records.

This is a convenience method equivalent to where().all()

Returns:

Type	Description
list[Model]	List of all model instances

Example

all_users = User.all()

Source code in packages/restmachine-orm/src/restmachine_orm/models/base.py

@classmethod
def all(cls) -> list["Model"]:
    """
    Get all records.

    This is a convenience method equivalent to where().all()

    Returns:
        List of all model instances

    Example:
        >>> all_users = User.all()
    """
    return cls.where().all()

Field

Field

Field(default: Any = PydanticUndefined, *, default_factory: Optional[Callable[[], Any]] = None, alias: Optional[str] = None, title: Optional[str] = None, description: Optional[str] = None, examples: Optional[list[Any]] = None, gt: Optional[float] = None, ge: Optional[float] = None, lt: Optional[float] = None, le: Optional[float] = None, min_length: Optional[int] = None, max_length: Optional[int] = None, pattern: Optional[str] = None, primary_key: bool = False, unique: bool = False, index: bool = False, searchable: bool = False, auto_now: bool = False, auto_now_add: bool = False, db_column: Optional[str] = None, gsi_partition_key: Optional[str] = None, gsi_sort_key: Optional[str] = None, analyzer: Optional[str] = None, **extra: Any) -> FieldInfo

Define a model field with validation and ORM metadata.

Parameters:

Name	Type	Description	Default
default	Any	Default value for the field	PydanticUndefined
default_factory	Optional[Callable[[], Any]]	Factory function for default values	None
alias	Optional[str]	Alternative name for the field	None
title	Optional[str]	Human-readable title	None
description	Optional[str]	Field description	None
examples	Optional[list[Any]]	Example values	None
gt	Optional[float]	Greater than validation	None
ge	Optional[float]	Greater than or equal validation	None
lt	Optional[float]	Less than validation	None
le	Optional[float]	Less than or equal validation	None
min_length	Optional[int]	Minimum string/list length	None
max_length	Optional[int]	Maximum string/list length	None
pattern	Optional[str]	Regex pattern for string validation	None
primary_key	bool	Whether this is the primary key	False
unique	bool	Whether values must be unique	False
index	bool	Whether to create a database index	False
searchable	bool	Whether to enable full-text search (OpenSearch)	False
auto_now	bool	Auto-update timestamp on save	False
auto_now_add	bool	Auto-set timestamp on create	False
db_column	Optional[str]	Custom database column name	None
gsi_partition_key	Optional[str]	DynamoDB GSI partition key name	None
gsi_sort_key	Optional[str]	DynamoDB GSI sort key name	None
analyzer	Optional[str]	OpenSearch text analyzer	None
**extra	Any	Additional Pydantic field arguments	{}

Returns:

Type	Description
FieldInfo	FieldInfo object with ORM metadata

Example

class User(Model): ... id: str = Field(primary_key=True) ... email: str = Field(unique=True, index=True) ... name: str = Field(max_length=100, searchable=True) ... age: int = Field(ge=0, le=150) ... created_at: datetime = Field(auto_now_add=True)

Decorators

partition_key

partition_key

partition_key(func: Callable[[Model], str]) -> Callable[[Model], str]

Decorator to mark a method as the partition key (hash key) generator.

Used for DynamoDB tables to define composite partition keys. The decorated method should return a string that will be used as the partition key value.

Parameters:

Name	Type	Description	Default
func	Callable[[Model], str]	Method that generates the partition key string	required

Returns:

Type	Description
Callable[[Model], str]	Decorated method with metadata

Example

class TodoItem(Model): ... user_id: str ... todo_id: str ... ... @partition_key ... def pk(self) -> str: ... return f"USER#{self.user_id}"

Note

Only one method per model should be decorated with @partition_key. The method name is conventionally 'pk' but can be anything.

Source code in packages/restmachine-orm/src/restmachine_orm/models/decorators.py

def partition_key(func: Callable[["Model"], str]) -> Callable[["Model"], str]:
    """
    Decorator to mark a method as the partition key (hash key) generator.

    Used for DynamoDB tables to define composite partition keys.
    The decorated method should return a string that will be used as the
    partition key value.

    Args:
        func: Method that generates the partition key string

    Returns:
        Decorated method with metadata

    Example:
        >>> class TodoItem(Model):
        ...     user_id: str
        ...     todo_id: str
        ...
        ...     @partition_key
        ...     def pk(self) -> str:
        ...         return f"USER#{self.user_id}"

    Note:
        Only one method per model should be decorated with @partition_key.
        The method name is conventionally 'pk' but can be anything.
    """
    @wraps(func)
    def wrapper(self: "Model") -> str:
        return func(self)

    # Mark the function with metadata
    wrapper._is_partition_key = True  # type: ignore
    wrapper._key_name = func.__name__  # type: ignore
    return wrapper

sort_key

sort_key

sort_key(func: Callable[[Model], str]) -> Callable[[Model], str]

Decorator to mark a method as the sort key (range key) generator.

Used for DynamoDB tables to define composite sort keys. The decorated method should return a string that will be used as the sort key value.

Parameters:

Name	Type	Description	Default
func	Callable[[Model], str]	Method that generates the sort key string	required

Returns:

Type	Description
Callable[[Model], str]	Decorated method with metadata

Example

class TodoItem(Model): ... user_id: str ... created_at: datetime ... todo_id: str ... ... @partition_key ... def pk(self) -> str: ... return f"USER#{self.user_id}" ... ... @sort_key ... def sk(self) -> str: ... return f"TODO#{self.created_at.isoformat()}#{self.todo_id}"

Note

Only one method per model should be decorated with @sort_key. The method name is conventionally 'sk' but can be anything.

Source code in packages/restmachine-orm/src/restmachine_orm/models/decorators.py

def sort_key(func: Callable[["Model"], str]) -> Callable[["Model"], str]:
    """
    Decorator to mark a method as the sort key (range key) generator.

    Used for DynamoDB tables to define composite sort keys.
    The decorated method should return a string that will be used as the
    sort key value.

    Args:
        func: Method that generates the sort key string

    Returns:
        Decorated method with metadata

    Example:
        >>> class TodoItem(Model):
        ...     user_id: str
        ...     created_at: datetime
        ...     todo_id: str
        ...
        ...     @partition_key
        ...     def pk(self) -> str:
        ...         return f"USER#{self.user_id}"
        ...
        ...     @sort_key
        ...     def sk(self) -> str:
        ...         return f"TODO#{self.created_at.isoformat()}#{self.todo_id}"

    Note:
        Only one method per model should be decorated with @sort_key.
        The method name is conventionally 'sk' but can be anything.
    """
    @wraps(func)
    def wrapper(self: "Model") -> str:
        return func(self)

    # Mark the function with metadata
    wrapper._is_sort_key = True  # type: ignore
    wrapper._key_name = func.__name__  # type: ignore
    return wrapper

DynamoDB-Specific Decorators

The following decorators are available for DynamoDB backend (in restmachine_orm.models.decorators):

• gsi_partition_key(index_name) - Mark method as GSI partition key generator
• gsi_sort_key(index_name) - Mark method as GSI sort key generator

These are used with the DynamoDB backend to define Global Secondary Index keys. See the DynamoDB Backend documentation for usage examples.