app.repository.base_repository module

class app.repository.base_repository.BaseRepository(session_factory, model)

Bases: object

Async base repository providing common CRUD operations on top of SQLAlchemy 2.0’s AsyncSession.

Parameters:

session_factory (Callable[[...], AbstractAsyncContextManager[AsyncSession]])

async read_by_options(schema, eager=False)

Run a filtered, ordered and paginated query from a search schema.

Non-null fields on schema are turned into WHERE conditions; ordering, page and page_size drive sorting and pagination (a page_size of "all" disables the limit). The total row count ignoring pagination is computed alongside the page.

Parameters:

• schema – Search schema whose non-null fields become filters and which may carry ordering/page/page_size.

• eager (bool) – When True, eagerly join the model’s eagers relationships.

Returns:

dict – {"items": [...], "search_options": {page, page_size, ordering, total_count}}.

async read_by_id(id, eager=False, not_found_raise_exception=True, not_found_message='Not found id : {id}')

Fetch a single row by primary key.

Parameters:

• id – Primary-key value to look up.

• eager (bool) – When True, eagerly join the model’s eagers relationships.

• not_found_raise_exception (bool) – Raise NotFoundError when no row matches; when False return None instead.

• not_found_message (str) – Format string for the not-found error ({id} is substituted).

Returns:

The matching model instance, or None when missing and not_found_raise_exception is False.

Raises:

NotFoundError – If no row matches and the flag is True.

async read_by_column(column, value, eager=False, only_one=True, not_found_raise_exception=True, not_found_message='Not found {column} : {value}')

Fetch rows whose column equals value.

Parameters:

• column (str) – Name of the model attribute to filter on.

• value – Value the column must equal.

• eager (bool) – When True, eagerly join the model’s eagers relationships.

• only_one (bool) – Return the first match when True; otherwise return the full list of matches.

• not_found_raise_exception (bool) – When only_one is True, raise NotFoundError if nothing matches.

• not_found_message (str) – Format string for the not-found error ({column} and {value} are substituted).

Returns:

A single instance (or None) when only_one is True, otherwise a list of matching instances.

Raises:

NotFoundError – If only_one and nothing matches and the flag is True.

async create(schema, session=None, auto_commit=True)

Insert a new row built from schema.

Supports two modes: a self-managed session (session=None, auto_commit=True) that opens, commits and closes its own session, or an externally-managed session passed by the caller so the insert can take part in a larger transaction (auto_commit=False flushes instead of committing).

Parameters:

• schema – Pydantic schema dumped into the model’s constructor.

• session (Optional[AsyncSession]) – Caller-managed session to reuse; a new one is opened when None.

• auto_commit (bool) – Commit immediately when True; otherwise only flush (requires an external session).

Returns:

The persisted (and refreshed) model instance.

Raises:

• ValueError – If auto_commit=False is requested without an external session.

• DuplicatedError – If the insert violates a uniqueness constraint.

async update(id, schema)

Partially update a row, ignoring None fields on schema.

Parameters:

• id – Primary key of the row to update.

• schema – Pydantic schema; only its non-null fields are written.

Returns:

The refreshed model instance after the update.

async update_attr(id, column, value)

Update a single column of a row.

Parameters:

• id – Primary key of the row to update.

• column (str) – Name of the column to set.

• value – New value for the column.

Returns:

The refreshed model instance after the update.

async whole_update(id, schema)

Fully replace a row’s columns from schema (including None).

Unlike update(), every field on schema is written, so this performs a complete overwrite rather than a partial patch.

Parameters:

• id – Primary key of the row to update.

• schema – Pydantic schema dumped in full into the update.

Returns:

The refreshed model instance after the update.

async delete_by_id(id)

Delete a row by primary key.

Parameters:

id – Primary key of the row to delete.

Raises:

NotFoundError – If no row matches id.

async read_by_columns(filters, eager=False, only_one=True, not_found_raise_exception=True)

Fetch rows matching several equality filters combined with AND.

Parameters:

• filters (dict) – Mapping of column name to required value; all conditions must hold (AND).

• eager (bool) – When True, eagerly join the model’s eagers relationships.

• only_one (bool) – Return the first match when True; otherwise return all matches.

• not_found_raise_exception (bool) – When only_one is True, raise NotFoundError if nothing matches.

Returns:

A single instance (or None) when only_one is True, otherwise a list of matching instances.

Raises:

NotFoundError – If only_one and nothing matches and the flag is True.