Skip to content

Backends API

Backend

Backend

Backend(adapter: ModelAdapter)

Bases: ABC

Abstract base class for storage backends.

All backends (DynamoDB, OpenSearch, Composite, etc.) must implement this interface to provide consistent CRUD operations.

Each backend has an adapter that knows how to map models to the backend's native storage format.

Backends support mixin extensions via the extension discovery system. Mixins can define inner classes that extend BackendExtension to provide backend-specific behavior.

Initialize backend with an adapter.

Parameters:

Name Type Description Default
adapter ModelAdapter

Adapter for mapping models to storage format

required
Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
def __init__(self, adapter: "ModelAdapter"):
    """
    Initialize backend with an adapter.

    Args:
        adapter: Adapter for mapping models to storage format
    """
    self.adapter = adapter
    self._configured_models: set[Type] = set()
    self._model_extensions: dict[Type, list['BackendExtension']] = {}

Attributes

backend_name abstractmethod property

backend_name: str

Backend identifier (e.g., 'dynamodb', 'memory', 'opensearch').

Used by extensions to determine if they should be activated.

Functions

create abstractmethod

create(model_class: type[Model], data: dict[str, Any]) -> dict[str, Any]

Create a new record in the backend.

Parameters:

Name Type Description Default
model_class type[Model]

The model class being created

required
data dict[str, Any]

Dictionary of field values

required

Returns:

Type Description
dict[str, Any]

Dictionary of created record data

Raises:

Type Description
ValidationError

If data validation fails

DuplicateKeyError

If unique constraint is violated

Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
@abstractmethod
def create(self, model_class: type["Model"], data: dict[str, Any]) -> dict[str, Any]:
    """
    Create a new record in the backend.

    Args:
        model_class: The model class being created
        data: Dictionary of field values

    Returns:
        Dictionary of created record data

    Raises:
        ValidationError: If data validation fails
        DuplicateKeyError: If unique constraint is violated
    """
    pass

upsert abstractmethod

upsert(model_class: type[Model], data: dict[str, Any]) -> dict[str, Any]

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.

Parameters:

Name Type Description Default
model_class type[Model]

The model class being upserted

required
data dict[str, Any]

Dictionary of field values

required

Returns:

Type Description
dict[str, Any]

Dictionary of upserted record data

Raises:

Type Description
ValidationError

If data validation fails

Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
@abstractmethod
def upsert(self, model_class: type["Model"], data: dict[str, Any]) -> dict[str, Any]:
    """
    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.

    Args:
        model_class: The model class being upserted
        data: Dictionary of field values

    Returns:
        Dictionary of upserted record data

    Raises:
        ValidationError: If data validation fails
    """
    pass

get abstractmethod

get(model_class: type[Model], **filters: Any) -> Optional[dict[str, Any]]

Get a single record by primary key or unique field.

Parameters:

Name Type Description Default
model_class type[Model]

The model class to query

required
**filters Any

Filter conditions (typically primary key)

{}

Returns:

Type Description
Optional[dict[str, Any]]

Dictionary of record data, or None if not found

Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
@abstractmethod
def get(self, model_class: type["Model"], **filters: Any) -> Optional[dict[str, Any]]:
    """
    Get a single record by primary key or unique field.

    Args:
        model_class: The model class to query
        **filters: Filter conditions (typically primary key)

    Returns:
        Dictionary of record data, or None if not found
    """
    pass

update abstractmethod

update(model_class: type[Model], instance: Model) -> dict[str, Any]

Update an existing record.

The backend is responsible for extracting keys from the instance using the model's metadata (e.g., partition key, sort key decorators).

Parameters:

Name Type Description Default
model_class type[Model]

The model class being updated

required
instance Model

The model instance with updated data

required

Returns:

Type Description
dict[str, Any]

Dictionary of updated record data

Raises:

Type Description
NotFoundError

If record doesn't exist

ValidationError

If update data is invalid

Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
@abstractmethod
def update(
    self,
    model_class: type["Model"],
    instance: "Model"
) -> dict[str, Any]:
    """
    Update an existing record.

    The backend is responsible for extracting keys from the instance
    using the model's metadata (e.g., partition key, sort key decorators).

    Args:
        model_class: The model class being updated
        instance: The model instance with updated data

    Returns:
        Dictionary of updated record data

    Raises:
        NotFoundError: If record doesn't exist
        ValidationError: If update data is invalid
    """
    pass

delete abstractmethod

delete(model_class: type[Model], instance: Model) -> bool

Delete a record.

The backend is responsible for extracting keys from the instance using the model's metadata (e.g., partition key, sort key decorators).

Parameters:

Name Type Description Default
model_class type[Model]

The model class being deleted

required
instance Model

The model instance to delete

required

Returns:

Type Description
bool

True if deleted, False if not found

Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
@abstractmethod
def delete(self, model_class: type["Model"], instance: "Model") -> bool:
    """
    Delete a record.

    The backend is responsible for extracting keys from the instance
    using the model's metadata (e.g., partition key, sort key decorators).

    Args:
        model_class: The model class being deleted
        instance: The model instance to delete

    Returns:
        True if deleted, False if not found
    """
    pass

query abstractmethod

query(model_class: type[Model]) -> QueryBuilder

Create a query builder for this backend.

Parameters:

Name Type Description Default
model_class type[Model]

The model class to query

required

Returns:

Type Description
QueryBuilder

QueryBuilder instance for fluent query construction

Example

backend.query(User).filter(age__gte=18).limit(10).all()

Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
@abstractmethod
def query(self, model_class: type["Model"]) -> "QueryBuilder":
    """
    Create a query builder for this backend.

    Args:
        model_class: The model class to query

    Returns:
        QueryBuilder instance for fluent query construction

    Example:
        >>> backend.query(User).filter(age__gte=18).limit(10).all()
    """
    pass

count abstractmethod

count(model_class: type[Model], **filters: Any) -> int

Count records matching filters.

Parameters:

Name Type Description Default
model_class type[Model]

The model class to count

required
**filters Any

Optional filter conditions

{}

Returns:

Type Description
int

Number of matching records

Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
@abstractmethod
def count(self, model_class: type["Model"], **filters: Any) -> int:
    """
    Count records matching filters.

    Args:
        model_class: The model class to count
        **filters: Optional filter conditions

    Returns:
        Number of matching records
    """
    pass

exists abstractmethod

exists(model_class: type[Model], **filters: Any) -> bool

Check if a record exists.

Parameters:

Name Type Description Default
model_class type[Model]

The model class to check

required
**filters Any

Filter conditions

{}

Returns:

Type Description
bool

True if at least one record matches, False otherwise

Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
@abstractmethod
def exists(self, model_class: type["Model"], **filters: Any) -> bool:
    """
    Check if a record exists.

    Args:
        model_class: The model class to check
        **filters: Filter conditions

    Returns:
        True if at least one record matches, False otherwise
    """
    pass

batch_create

batch_create(model_class: type[Model], items: list[dict[str, Any]]) -> list[dict[str, Any]]

Create multiple records in a single batch operation.

Default implementation calls create() for each item. Backends should override with optimized batch operations.

Parameters:

Name Type Description Default
model_class type[Model]

The model class being created

required
items list[dict[str, Any]]

List of record data dictionaries

required

Returns:

Type Description
list[dict[str, Any]]

List of created record data

Raises:

Type Description
ValidationError

If any item validation fails

Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
def batch_create(
    self,
    model_class: type["Model"],
    items: list[dict[str, Any]]
) -> list[dict[str, Any]]:
    """
    Create multiple records in a single batch operation.

    Default implementation calls create() for each item.
    Backends should override with optimized batch operations.

    Args:
        model_class: The model class being created
        items: List of record data dictionaries

    Returns:
        List of created record data

    Raises:
        ValidationError: If any item validation fails
    """
    return [self.create(model_class, item) for item in items]

batch_get

batch_get(model_class: type[Model], keys: list[dict[str, Any]]) -> list[Optional[dict[str, Any]]]

Get multiple records by their keys in a single batch operation.

Default implementation calls get() for each key. Backends should override with optimized batch operations.

Parameters:

Name Type Description Default
model_class type[Model]

The model class to query

required
keys list[dict[str, Any]]

List of key dictionaries

required

Returns:

Type Description
list[Optional[dict[str, Any]]]

List of record data (None for missing records)

Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
def batch_get(
    self,
    model_class: type["Model"],
    keys: list[dict[str, Any]]
) -> list[Optional[dict[str, Any]]]:
    """
    Get multiple records by their keys in a single batch operation.

    Default implementation calls get() for each key.
    Backends should override with optimized batch operations.

    Args:
        model_class: The model class to query
        keys: List of key dictionaries

    Returns:
        List of record data (None for missing records)
    """
    return [self.get(model_class, **key) for key in keys]

initialize

initialize() -> None

Initialize the backend (create tables, indexes, etc.).

Optional method for backends that need setup.

Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
def initialize(self) -> None:
    """
    Initialize the backend (create tables, indexes, etc.).

    Optional method for backends that need setup.
    """
    pass

close

close() -> None

Close connections and cleanup resources.

Optional method for backends with persistent connections.

Source code in packages/restmachine-orm/src/restmachine_orm/backends/base.py
def close(self) -> None:
    """
    Close connections and cleanup resources.

    Optional method for backends with persistent connections.
    """
    pass

InMemoryBackend

InMemoryBackend

InMemoryBackend(adapter: Optional[InMemoryAdapter] = None)

Bases: Backend

In-memory storage backend using Python dicts.

Stores all data in memory. Data is lost when the process ends. Useful for testing and examples.

Example

from restmachine_orm.backends.memory import InMemoryBackend from restmachine_orm.backends.adapters import InMemoryAdapter

class User(Model): ... class Meta: ... backend = InMemoryBackend(InMemoryAdapter()) ... id: str = Field(primary_key=True) ... name: str

user = User.create(id="123", name="Alice")

Initialize in-memory backend.

Parameters:

Name Type Description Default
adapter Optional[InMemoryAdapter]

Optional adapter (uses InMemoryAdapter by default)

None
Source code in packages/restmachine-orm/src/restmachine_orm/backends/memory.py
def __init__(self, adapter: Optional[InMemoryAdapter] = None):
    """
    Initialize in-memory backend.

    Args:
        adapter: Optional adapter (uses InMemoryAdapter by default)
    """
    super().__init__(adapter or InMemoryAdapter())
    # Storage: {model_class_name: {pk: record_data}}
    self._storage: dict[str, dict[Any, dict[str, Any]]] = {}

Attributes

backend_name property

backend_name: str

Backend identifier.

Functions

create

create(model_class: type[Model], data: dict[str, Any]) -> dict[str, Any]

Create a new record in memory.

Source code in packages/restmachine-orm/src/restmachine_orm/backends/memory.py
def create(self, model_class: type["Model"], data: dict[str, Any]) -> dict[str, Any]:
    """Create a new record in memory."""
    # Ensure extensions are configured
    self._ensure_configured(model_class)

    storage = self._get_storage(model_class)

    # Create model instance to get primary key
    instance = model_class(**data)
    pk_value = self.adapter.get_primary_key_value(instance)

    # Check if already exists
    if pk_value in storage:
        from restmachine_orm.backends.base import DuplicateKeyError
        raise DuplicateKeyError(f"Record with key {pk_value} already exists")

    # Convert to storage format
    storage_data = self.adapter.model_to_storage(instance)

    # Call extension hooks
    storage_data = self._call_serialize_hooks(model_class, storage_data)
    self._call_validate_hooks(model_class, storage_data)

    # Store data
    storage[pk_value] = deepcopy(storage_data)

    # Return model format (apply deserialize hooks)
    result_data = self.adapter.storage_to_model(model_class, storage_data)
    result_data = self._call_deserialize_hooks(model_class, result_data)
    return result_data

upsert

upsert(model_class: type[Model], data: dict[str, Any]) -> dict[str, Any]

Create or update a record (upsert).

Source code in packages/restmachine-orm/src/restmachine_orm/backends/memory.py
def upsert(self, model_class: type["Model"], data: dict[str, Any]) -> dict[str, Any]:
    """Create or update a record (upsert)."""
    # Ensure extensions are configured
    self._ensure_configured(model_class)

    storage = self._get_storage(model_class)

    # Create model instance to get primary key
    instance = model_class(**data)
    pk_value = self.adapter.get_primary_key_value(instance)

    # Convert to storage format
    storage_data = self.adapter.model_to_storage(instance)

    # Call extension hooks
    storage_data = self._call_serialize_hooks(model_class, storage_data)
    self._call_validate_hooks(model_class, storage_data)

    # Store data (overwrite if exists)
    storage[pk_value] = deepcopy(storage_data)

    # Return model format (apply deserialize hooks)
    result_data = self.adapter.storage_to_model(model_class, storage_data)
    result_data = self._call_deserialize_hooks(model_class, result_data)
    return result_data

get

get(model_class: type[Model], **filters: Any) -> Optional[dict[str, Any]]

Get a single record by primary key.

Source code in packages/restmachine-orm/src/restmachine_orm/backends/memory.py
def get(self, model_class: type["Model"], **filters: Any) -> Optional[dict[str, Any]]:
    """Get a single record by primary key."""
    storage = self._get_storage(model_class)

    # Simple implementation: assumes filters contain primary key
    # For more complex filtering, use query()
    for record in storage.values():
        model_data = self.adapter.storage_to_model(model_class, record)
        model_data = self._call_deserialize_hooks(model_class, model_data)
        if all(model_data.get(k) == v for k, v in filters.items()):
            return model_data

    return None

update

update(model_class: type[Model], instance: Model) -> dict[str, Any]

Update an existing record.

Source code in packages/restmachine-orm/src/restmachine_orm/backends/memory.py
def update(
    self,
    model_class: type["Model"],
    instance: "Model"
) -> dict[str, Any]:
    """Update an existing record."""
    storage = self._get_storage(model_class)

    # Get primary key from instance
    pk_value = self.adapter.get_primary_key_value(instance)

    if pk_value not in storage:
        raise NotFoundError(f"Record not found with key: {pk_value}")

    # Convert to storage format
    storage_data = self.adapter.model_to_storage(instance)

    # Call extension hooks
    storage_data = self._call_serialize_hooks(model_class, storage_data)
    self._call_validate_hooks(model_class, storage_data)

    # Update the record
    storage[pk_value] = deepcopy(storage_data)

    # Return model format (apply deserialize hooks)
    # Note: For update, this return value is currently ignored by Model.save()
    # but we return it for consistency and future hooks support
    result_data = self.adapter.storage_to_model(model_class, storage_data)
    result_data = self._call_deserialize_hooks(model_class, result_data)
    return result_data

delete

delete(model_class: type[Model], instance: Model) -> bool

Delete a record.

Source code in packages/restmachine-orm/src/restmachine_orm/backends/memory.py
def delete(self, model_class: type["Model"], instance: "Model") -> bool:
    """Delete a record."""
    storage = self._get_storage(model_class)

    # Get primary key from instance
    pk_value = self.adapter.get_primary_key_value(instance)

    if pk_value in storage:
        del storage[pk_value]
        return True

    return False

query

query(model_class: type[Model]) -> QueryBuilder

Create a query builder for this backend.

Source code in packages/restmachine-orm/src/restmachine_orm/backends/memory.py
def query(self, model_class: type["Model"]) -> QueryBuilder:
    """Create a query builder for this backend."""
    query_builder: QueryBuilder = InMemoryQueryBuilder(model_class, self)
    # Call extension hooks to modify query behavior
    query_builder = self._call_modify_query_hooks(model_class, query_builder)
    return query_builder

count

count(model_class: type[Model], **filters: Any) -> int

Count records matching filters.

Source code in packages/restmachine-orm/src/restmachine_orm/backends/memory.py
def count(self, model_class: type["Model"], **filters: Any) -> int:
    """Count records matching filters."""
    storage = self._get_storage(model_class)

    if not filters:
        return len(storage)

    count = 0
    for record in storage.values():
        model_data = self.adapter.storage_to_model(model_class, record)
        if all(model_data.get(k) == v for k, v in filters.items()):
            count += 1

    return count

exists

exists(model_class: type[Model], **filters: Any) -> bool

Check if a record exists.

Source code in packages/restmachine-orm/src/restmachine_orm/backends/memory.py
def exists(self, model_class: type["Model"], **filters: Any) -> bool:
    """Check if a record exists."""
    return self.count(model_class, **filters) > 0

clear

clear(model_class: Optional[type[Model]] = None) -> None

Clear storage.

Parameters:

Name Type Description Default
model_class Optional[type[Model]]

Optional model class to clear. If None, clears all.

None
Source code in packages/restmachine-orm/src/restmachine_orm/backends/memory.py
def clear(self, model_class: Optional[type["Model"]] = None) -> None:
    """
    Clear storage.

    Args:
        model_class: Optional model class to clear. If None, clears all.
    """
    if model_class:
        entity_type = self.adapter.get_entity_type(model_class)
        self._storage.pop(entity_type, None)
    else:
        self._storage.clear()

InMemoryAdapter

InMemoryAdapter

Bases: ModelAdapter

Adapter for in-memory storage (testing and examples).

Simple adapter that uses the primary key field as-is. No special transformation needed for in-memory dicts.

Functions

model_to_storage

model_to_storage(instance: Model) -> dict[str, Any]

Return model data as-is for in-memory storage.

Source code in packages/restmachine-orm/src/restmachine_orm/backends/adapters.py
def model_to_storage(self, instance: "Model") -> dict[str, Any]:
    """Return model data as-is for in-memory storage."""
    return instance.model_dump()

storage_to_model

storage_to_model(model_class: type[Model], data: dict[str, Any]) -> dict[str, Any]

Return data as-is for model instantiation.

Source code in packages/restmachine-orm/src/restmachine_orm/backends/adapters.py
def storage_to_model(
    self,
    model_class: type["Model"],
    data: dict[str, Any]
) -> dict[str, Any]:
    """Return data as-is for model instantiation."""
    return dict(data)

get_primary_key_value

get_primary_key_value(instance: Model) -> Any

Get primary key value.

Source code in packages/restmachine-orm/src/restmachine_orm/backends/adapters.py
def get_primary_key_value(self, instance: "Model") -> Any:
    """Get primary key value."""
    from restmachine_orm.models.fields import get_field_orm_metadata

    # Find the primary key field
    pk_field = None
    for field_name, field_info in instance.__class__.model_fields.items():
        metadata = get_field_orm_metadata(field_info)
        if metadata.get("primary_key"):
            pk_field = field_name
            break

    if not pk_field:
        raise ValueError(
            f"No primary key field defined for {instance.__class__.__name__}"
        )
    return getattr(instance, pk_field)

ModelAdapter

ModelAdapter

Bases: ABC

Abstract adapter for mapping models to storage backends.

Each backend implements its own adapter that understands how to translate between the generic Model representation and the backend-specific storage format.

Functions

model_to_storage abstractmethod

model_to_storage(instance: Model) -> dict[str, Any]

Transform a model instance to backend storage format.

Parameters:

Name Type Description Default
instance Model

Model instance to transform

required

Returns:

Type Description
dict[str, Any]

Dictionary in backend-specific format

Example (DynamoDB): >>> adapter.model_to_storage(user) { 'pk': 'USER#123', 'sk': 'METADATA', 'entity_type': 'User', 'id': '123', 'email': 'alice@example.com', 'name': 'Alice' }

Source code in packages/restmachine-orm/src/restmachine_orm/backends/adapters.py
@abstractmethod
def model_to_storage(self, instance: "Model") -> dict[str, Any]:
    """
    Transform a model instance to backend storage format.

    Args:
        instance: Model instance to transform

    Returns:
        Dictionary in backend-specific format

    Example (DynamoDB):
        >>> adapter.model_to_storage(user)
        {
            'pk': 'USER#123',
            'sk': 'METADATA',
            'entity_type': 'User',
            'id': '123',
            'email': 'alice@example.com',
            'name': 'Alice'
        }
    """
    pass

storage_to_model abstractmethod

storage_to_model(model_class: type[Model], data: dict[str, Any]) -> dict[str, Any]

Transform backend storage format to model data.

Parameters:

Name Type Description Default
model_class type[Model]

The model class to instantiate

required
data dict[str, Any]

Data from backend storage

required

Returns:

Type Description
dict[str, Any]

Dictionary suitable for model instantiation

Example (DynamoDB): >>> adapter.storage_to_model(User, { ... 'pk': 'USER#123', ... 'sk': 'METADATA', ... 'entity_type': 'User', ... 'id': '123', ... 'email': 'alice@example.com' ... })

Source code in packages/restmachine-orm/src/restmachine_orm/backends/adapters.py
@abstractmethod
def storage_to_model(
    self,
    model_class: type["Model"],
    data: dict[str, Any]
) -> dict[str, Any]:
    """
    Transform backend storage format to model data.

    Args:
        model_class: The model class to instantiate
        data: Data from backend storage

    Returns:
        Dictionary suitable for model instantiation

    Example (DynamoDB):
        >>> adapter.storage_to_model(User, {
        ...     'pk': 'USER#123',
        ...     'sk': 'METADATA',
        ...     'entity_type': 'User',
        ...     'id': '123',
        ...     'email': 'alice@example.com'
        ... })
        {'id': '123', 'email': 'alice@example.com'}
    """
    pass

get_primary_key_value abstractmethod

get_primary_key_value(instance: Model) -> Any

Get the primary key value for a model instance.

For simple backends (like SQL), this is just the primary key field. For DynamoDB, this might be a composite key.

Parameters:

Name Type Description Default
instance Model

Model instance

required

Returns:

Type Description
Any

Primary key value (simple or composite)

Example

adapter.get_primary_key_value(user) '123' # Simple key adapter.get_primary_key_value(todo) {'pk': 'USER#alice', 'sk': 'TODO#2025-01-15'} # Composite

Source code in packages/restmachine-orm/src/restmachine_orm/backends/adapters.py
@abstractmethod
def get_primary_key_value(self, instance: "Model") -> Any:
    """
    Get the primary key value for a model instance.

    For simple backends (like SQL), this is just the primary key field.
    For DynamoDB, this might be a composite key.

    Args:
        instance: Model instance

    Returns:
        Primary key value (simple or composite)

    Example:
        >>> adapter.get_primary_key_value(user)
        '123'  # Simple key
        >>> adapter.get_primary_key_value(todo)
        {'pk': 'USER#alice', 'sk': 'TODO#2025-01-15'}  # Composite
    """
    pass

get_index_keys

get_index_keys(instance: Model) -> dict[str, Any]

Get secondary index keys for a model instance.

Optional method for backends that support secondary indexes.

Parameters:

Name Type Description Default
instance Model

Model instance

required

Returns:

Type Description
dict[str, Any]

Dictionary of index names to key values

Example (DynamoDB GSIs): >>> adapter.get_index_keys(user) { 'EmailIndex': {'gsi_pk': 'alice@example.com'}, 'TenantIndex': {'gsi_pk': 'TENANT#org-1', 'gsi_sk': '2025-01-15'} }

Source code in packages/restmachine-orm/src/restmachine_orm/backends/adapters.py
def get_index_keys(self, instance: "Model") -> dict[str, Any]:
    """
    Get secondary index keys for a model instance.

    Optional method for backends that support secondary indexes.

    Args:
        instance: Model instance

    Returns:
        Dictionary of index names to key values

    Example (DynamoDB GSIs):
        >>> adapter.get_index_keys(user)
        {
            'EmailIndex': {'gsi_pk': 'alice@example.com'},
            'TenantIndex': {'gsi_pk': 'TENANT#org-1', 'gsi_sk': '2025-01-15'}
        }
    """
    return {}

get_entity_type

get_entity_type(model_class: type[Model]) -> str

Get the entity type identifier for a model.

Used for filtering in single-table designs.

Parameters:

Name Type Description Default
model_class type[Model]

Model class

required

Returns:

Type Description
str

Entity type identifier (typically the class name)

Source code in packages/restmachine-orm/src/restmachine_orm/backends/adapters.py
def get_entity_type(self, model_class: type["Model"]) -> str:
    """
    Get the entity type identifier for a model.

    Used for filtering in single-table designs.

    Args:
        model_class: Model class

    Returns:
        Entity type identifier (typically the class name)
    """
    return model_class.__name__

Exceptions

NotFoundError

NotFoundError

Bases: BackendError

Record not found in backend.

DuplicateKeyError

DuplicateKeyError

Bases: BackendError

Unique constraint violation.