abc

class litestar.repository.abc.AbstractAsyncRepository

Bases: Generic[T]

Interface for persistent data interaction.

model_type: type[T]

Type of object represented by the repository.

id_attribute: Any = 'id'

Name of the primary identifying attribute on model_type.

__init__(**kwargs: Any) → None

Repository constructors accept arbitrary kwargs.

abstract async add(data: T) → T

Add data to the collection.

Parameters:

data – Instance to be added to the collection.

Returns:

The added instance.

abstract async add_many(data: list[T]) → list[T]

Add multiple data to the collection.

Parameters:

data – Instances to be added to the collection.

Returns:

The added instances.

abstract async count(*filters: FilterTypes, **kwargs: Any) → int

Get the count of records returned by a query.

Parameters:

• *filters – Types for specific filtering operations.

• **kwargs – Instance attribute value filters.

Returns:

The count of instances

abstract async delete(item_id: Any) → T

Delete instance identified by item_id.

Parameters:

item_id – Identifier of instance to be deleted.

Returns:

The deleted instance.

Raises:

NotFoundError – If no instance found identified by item_id.

abstract async delete_many(item_ids: list[Any]) → list[T]

Delete multiple instances identified by list of IDs item_ids.

Parameters:

item_ids – list of Identifiers to be deleted.

Returns:

The deleted instances.

abstract async exists(*filters: FilterTypes, **kwargs: Any) → bool

Return true if the object specified by kwargs exists.

Parameters:

• *filters – Types for specific filtering operations.

• **kwargs – Identifier of the instance to be retrieved.

Returns:

True if the instance was found. False if not found.

abstract async get(item_id: Any, **kwargs: Any) → T

Get instance identified by item_id.

Parameters:

• item_id – Identifier of the instance to be retrieved.

• **kwargs – Additional arguments

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

abstract async get_one(**kwargs: Any) → T

Get an instance specified by the kwargs filters if it exists.

Parameters:

**kwargs – Instance attribute value filters.

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by kwargs.

abstract async get_or_create(**kwargs: Any) → tuple[T, bool]

Get an instance specified by the kwargs filters if it exists or create it.

Parameters:

**kwargs – Instance attribute value filters.

Returns:

A tuple that includes the retrieved or created instance, and a boolean on whether the record was created or not

abstract async get_one_or_none(**kwargs: Any) → T | None

Get an instance if it exists or None.

Parameters:

**kwargs – Instance attribute value filters.

Returns:

The retrieved instance or None.

abstract async update(data: T) → T

Update instance with the attribute values present on data.

Parameters:

data – An instance that should have a value for id_attribute that exists in the collection.

Returns:

The updated instance.

Raises:

NotFoundError – If no instance found with same identifier as data.

abstract async update_many(data: list[T]) → list[T]

Update multiple instances with the attribute values present on instances in data.

Parameters:

data – A list of instance that should have a value for id_attribute that exists in the collection.

Returns:

a list of the updated instances.

Raises:

NotFoundError – If no instance found with same identifier as data.

abstract async upsert(data: T) → T

Update or create instance.

Updates instance with the attribute values present on data, or creates a new instance if one doesn’t exist.

Parameters:

data – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of id_attribute.

Returns:

The updated or created instance.

Raises:

NotFoundError – If no instance found with same identifier as data.

abstract async upsert_many(data: list[T]) → list[T]

Update or create multiple instances.

Update instances with the attribute values present on data, or create a new instance if one doesn’t exist.

Parameters:

data – Instances to update or created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of id_attribute.

Returns:

The updated or created instances.

Raises:

NotFoundError – If no instance found with same identifier as data.

abstract async list_and_count(*filters: FilterTypes, **kwargs: Any) → tuple[list[T], int]

List records with total count.

Parameters:

• *filters – Types for specific filtering operations.

• **kwargs – Instance attribute value filters.

Returns:

a tuple containing The list of instances, after filtering applied, and a count of records returned by query, ignoring pagination.

abstract async list(*filters: FilterTypes, **kwargs: Any) → list[T]

Get a list of instances, optionally filtered.

Parameters:

• *filters – filters for specific filtering operations

• **kwargs – Instance attribute value filters.

Returns:

The list of instances, after filtering applied

abstract filter_collection_by_kwargs(collection: CollectionT, /, **kwargs: Any) → CollectionT

Filter the collection by kwargs.

Has AND semantics where multiple kwargs name/value pairs are provided.

Parameters:

• collection – the objects to be filtered

• **kwargs – key/value pairs such that objects remaining in the collection after filtering have the property that their attribute named key has value equal to value.

Returns:

The filtered objects

Raises:

RepositoryError – if a named attribute doesn’t exist on model_type.

static check_not_found(item_or_none: T | None) → T

Raise NotFoundError if item_or_none is None.

Parameters:

item_or_none – Item (T) to be tested for existence.

Returns:

The item, if it exists.

classmethod get_id_attribute_value(item: T | type[T], id_attribute: str | None = None) → Any

Get value of attribute named as id_attribute on item.

Parameters:

• item – Anything that should have an attribute named as id_attribute value.

• id_attribute – Allows customization of the unique identifier to use for model fetching. Defaults to None, but can reference any surrogate or candidate key for the table.

Returns:

The value of attribute on item named as id_attribute.

classmethod set_id_attribute_value(item_id: Any, item: T, id_attribute: str | None = None) → T

Return the item after the ID is set to the appropriate attribute.

Parameters:

• item_id – Value of ID to be set on instance

• item – Anything that should have an attribute named as id_attribute value.

• id_attribute – Allows customization of the unique identifier to use for model fetching. Defaults to None, but can reference any surrogate or candidate key for the table.

Returns:

Item with item_id set to id_attribute

class litestar.repository.abc.AbstractSyncRepository

Bases: Generic[T]

Interface for persistent data interaction.

model_type: type[T]

Type of object represented by the repository.

id_attribute: Any = 'id'

Name of the primary identifying attribute on model_type.

__init__(**kwargs: Any) → None

Repository constructors accept arbitrary kwargs.

abstract add(data: T) → T

Add data to the collection.

Parameters:

data – Instance to be added to the collection.

Returns:

The added instance.

abstract add_many(data: list[T]) → list[T]

Add multiple data to the collection.

Parameters:

data – Instances to be added to the collection.

Returns:

The added instances.

abstract count(*filters: FilterTypes, **kwargs: Any) → int

Get the count of records returned by a query.

Parameters:

• *filters – Types for specific filtering operations.

• **kwargs – Instance attribute value filters.

Returns:

The count of instances

abstract delete(item_id: Any) → T

Delete instance identified by item_id.

Parameters:

item_id – Identifier of instance to be deleted.

Returns:

The deleted instance.

Raises:

NotFoundError – If no instance found identified by item_id.

abstract delete_many(item_ids: list[Any]) → list[T]

Delete multiple instances identified by list of IDs item_ids.

Parameters:

item_ids – list of Identifiers to be deleted.

Returns:

The deleted instances.

abstract exists(*filters: FilterTypes, **kwargs: Any) → bool

Return true if the object specified by kwargs exists.

Parameters:

• *filters – Types for specific filtering operations.

• **kwargs – Identifier of the instance to be retrieved.

Returns:

True if the instance was found. False if not found.

abstract get(item_id: Any, **kwargs: Any) → T

Get instance identified by item_id.

Parameters:

• item_id – Identifier of the instance to be retrieved.

• **kwargs – Additional arguments

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

abstract get_one(**kwargs: Any) → T

Get an instance specified by the kwargs filters if it exists.

Parameters:

**kwargs – Instance attribute value filters.

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by kwargs.

abstract get_or_create(**kwargs: Any) → tuple[T, bool]

Get an instance specified by the kwargs filters if it exists or create it.

Parameters:

**kwargs – Instance attribute value filters.

Returns:

A tuple that includes the retrieved or created instance, and a boolean on whether the record was created or not

abstract get_one_or_none(**kwargs: Any) → T | None

Get an instance if it exists or None.

Parameters:

**kwargs – Instance attribute value filters.

Returns:

The retrieved instance or None.

abstract update(data: T) → T

Update instance with the attribute values present on data.

Parameters:

data – An instance that should have a value for id_attribute that exists in the collection.

Returns:

The updated instance.

Raises:

NotFoundError – If no instance found with same identifier as data.

abstract update_many(data: list[T]) → list[T]

Update multiple instances with the attribute values present on instances in data.

Parameters:

data – A list of instance that should have a value for id_attribute that exists in the collection.

Returns:

a list of the updated instances.

Raises:

NotFoundError – If no instance found with same identifier as data.

abstract upsert(data: T) → T

Update or create instance.

Updates instance with the attribute values present on data, or creates a new instance if one doesn’t exist.

Parameters:

data – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of id_attribute.

Returns:

The updated or created instance.

Raises:

NotFoundError – If no instance found with same identifier as data.

abstract upsert_many(data: list[T]) → list[T]

Update or create multiple instances.

Update instances with the attribute values present on data, or create a new instance if one doesn’t exist.

Parameters:

data – Instances to update or created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of id_attribute.

Returns:

The updated or created instances.

Raises:

NotFoundError – If no instance found with same identifier as data.

abstract list_and_count(*filters: FilterTypes, **kwargs: Any) → tuple[list[T], int]

List records with total count.

Parameters:

• *filters – Types for specific filtering operations.

• **kwargs – Instance attribute value filters.

Returns:

a tuple containing The list of instances, after filtering applied, and a count of records returned by query, ignoring pagination.

abstract list(*filters: FilterTypes, **kwargs: Any) → list[T]

Get a list of instances, optionally filtered.

Parameters:

• *filters – filters for specific filtering operations

• **kwargs – Instance attribute value filters.

Returns:

The list of instances, after filtering applied

abstract filter_collection_by_kwargs(collection: CollectionT, /, **kwargs: Any) → CollectionT

Filter the collection by kwargs.

Has AND semantics where multiple kwargs name/value pairs are provided.

Parameters:

• collection – the objects to be filtered

• **kwargs – key/value pairs such that objects remaining in the collection after filtering have the property that their attribute named key has value equal to value.

Returns:

The filtered objects

Raises:

RepositoryError – if a named attribute doesn’t exist on model_type.

static check_not_found(item_or_none: T | None) → T

Raise NotFoundError if item_or_none is None.

Parameters:

item_or_none – Item (T) to be tested for existence.

Returns:

The item, if it exists.

classmethod get_id_attribute_value(item: T | type[T], id_attribute: str | None = None) → Any

Get value of attribute named as id_attribute on item.

Parameters:

• item – Anything that should have an attribute named as id_attribute value.

• id_attribute – Allows customization of the unique identifier to use for model fetching. Defaults to None, but can reference any surrogate or candidate key for the table.

Returns:

The value of attribute on item named as id_attribute.

classmethod set_id_attribute_value(item_id: Any, item: T, id_attribute: str | None = None) → T

Return the item after the ID is set to the appropriate attribute.

Parameters:

• item_id – Value of ID to be set on instance

• item – Anything that should have an attribute named as id_attribute value.

• id_attribute – Allows customization of the unique identifier to use for model fetching. Defaults to None, but can reference any surrogate or candidate key for the table.

Returns:

Item with item_id set to id_attribute