dbrepos.core package

Submodules

dbrepos.core.abstract module

class dbrepos.core.abstract.IFilter(table_class: Type[TTableCovariant], column_name: str, value: TFieldValue | None = None, operator_: operator = operator.eq)[source]

Bases: Protocol[TTableCovariant, TColumn, TFieldValue, TCompiledFilter]

__call__(value: TFieldValue, operator_: operator = operator.eq) → Self[source]

Finish construction of the object

This is a 2nd-step construction for DI support

Parameters:

  • value (TFieldValue) – Value to filter against

  • operator (operator, optional) – Operator for filtering. Defaults to operator.eq

Returns:

Completed filter object

Return type:

IFilter

Examples

` uuid_filter = Filter(File, "uuid")([str(uuid4()), str(uuid4())], operator.in_)  # noqa:E501 `

column: TColumn
column_name: str
compile() → TCompiledFilter[source]

Compile filter

Returns:

Compiled filter for usage in orm or IFilterSeq class

Return type:

TCompiledFilter

operator_: operator
value: TFieldValue | None
class dbrepos.core.abstract.IFilterSeq(mode_: mode, *filters: IFilter | IFilterSeq[TCompiledFilter])[source]

Bases: Protocol[TCompiledFilter]

compile() → TCompiledFilter[source]

Compile Filter Sequence

Returns:

Compiled filter for usage in orm

Return type:

TCompiledFilter

class dbrepos.core.abstract.IRepo(*, table_class: Type[TTable], pk_field_name: str = 'id', is_soft_deletable: bool = False, default_ordering: Tuple[str, ...] = ('id',), session_factory: AbstractContextManager | None = None)[source]

Bases: Protocol[TTable, TResultORM]

all(*, extra: Extra | None = None, session: TSession | None = None) → Iterable[TResultORM]
all(*, convert_to: Type[TResultDataclass], extra: Extra | None = None, session: TSession | None = None) → Iterable[TResultDataclass]

Helper for @overload to raise when called.

all_by_field(*, name: str, value: TFieldValue, extra: Extra | None = None, session: TSession | None = None) → Iterable[TResultORM]
all_by_field(*, name: str, value: TFieldValue, convert_to: Type[TResultDataclass], extra: Extra | None = None, session: TSession | None = None) → Iterable[TResultDataclass]

Helper for @overload to raise when called.

all_by_filters(*, filters: IFilterSeq, extra: Extra | None = None, session: TSession | None = None) → Iterable[TResultORM]
all_by_filters(*, filters: IFilterSeq, convert_to: Type[TResultDataclass], extra: Extra | None = None, session: TSession | None = None) → Iterable[TResultDataclass]

Helper for @overload to raise when called.

all_by_pks(pks: Sequence[TPrimaryKey], *, extra: Extra | None = None, session: TSession | None = None) → Iterable[TResultORM]
all_by_pks(pks: Sequence[TPrimaryKey], *, convert_to: Type[TResultDataclass], extra: Extra | None = None, session: TSession | None = None) → Iterable[TResultDataclass]

Helper for @overload to raise when called.

count_by_field(*, name: str, value: TFieldValue, extra: Extra | None = None, session: TSession | None = None) → int[source]

Count rows by field:value

Parameters:

  • name (str) – Name of the field

  • value (TFieldValue) – Value of the field

  • extra (Extra | None, optional) – Extra params. Defaults to None

  • session (TSession | None) – Session to use for DB queries. Defaults to None. Currently supported for SQLAlchemy

Returns:

Number of found rows

Return type:

int

count_by_filters(*, filters: IFilterSeq, extra: Extra | None = None, session: TSession | None = None) → int[source]

Count rows by filters

Parameters:

  • filters (IFilterSeq) – Filter sequence

  • extra (Extra | None, optional) – Extra params. Defaults to None

  • session (TSession | None) – Session to use for DB queries. Defaults to None. Currently supported for SQLAlchemy

Returns:

Number of found rows

Return type:

int

create(entity: TEntity, *, session: TSession | None = None) → TResultORM
create(entity: TEntity, *, convert_to: Type[TResultDataclass], session: TSession | None = None) → TResultDataclass

Helper for @overload to raise when called.

default_ordering: Tuple[str, ...]
delete(pk: TPrimaryKey, *, extra: Extra | None = None, session: TSession | None = None) → None[source]

Delete row by pk

Parameters:

  • pk (TPrimaryKey) – Primary key of row to delete

  • extra (Extra | None, optional) – Extra params. Defaults to None

  • session (TSession | None) – Session to use for DB queries. Defaults to None. Currently supported for SQLAlchemy

delete_by_field(*, name: str, value: TFieldValue, extra: Extra | None = None, session: TSession | None = None) → None[source]

Delete row by field:value

Parameters:

  • name (str) – Name of the field

  • value (TFieldValue) – Value of the field

  • extra (Extra | None, optional) – Extra params. Defaults to None

  • session (TSession | None) – Session to use for DB queries. Defaults to None. Currently supported for SQLAlchemy

exists_by_field(*, name: str, value: TFieldValue, extra: Extra | None = None, session: TSession | None = None) → bool[source]

Check if row exists by field:value

Parameters:

  • name (str) – Name of the field

  • value (TFieldValue) – Value of the field

  • extra (Extra | None, optional) – Extra params. Defaults to None

  • session (TSession | None) – Session to use for DB queries. Defaults to None. Currently supported for SQLAlchemy

Returns:

Row existence

Return type:

bool

exists_by_filters(*, filters: IFilterSeq, extra: Extra | None = None, session: TSession | None = None) → bool[source]

Check if row exists by filters

Parameters:

  • filters (IFilterSeq) – Filter sequence

  • extra (Extra | None, optional) – Extra params. Defaults to None

  • session (TSession | None) – Session to use for DB queries. Defaults to None. Currently supported for SQLAlchemy

Returns:

Row existence

Return type:

bool

get_by_field(*, name: str, value: TFieldValue, strict: Literal[True] = True, extra: Extra | None = None, session: TSession | None = None) → TResultORM
get_by_field(*, name: str, value: TFieldValue, convert_to: Type[TResultDataclass], strict: Literal[True] = True, extra: Extra | None = None, session: TSession | None = None) → TResultDataclass
get_by_field(*, name: str, value: TFieldValue, strict: Literal[False], extra: Extra | None = None, session: TSession | None = None) → TResultORM | None
get_by_field(*, name: str, value: TFieldValue, convert_to: Type[TResultDataclass], strict: Literal[False], extra: Extra | None = None, session: TSession | None = None) → TResultDataclass | None

Helper for @overload to raise when called.

get_by_filters(*, filters: IFilterSeq, strict: Literal[True] = True, extra: Extra | None = None, session: TSession | None = None) → TResultORM
get_by_filters(*, filters: IFilterSeq, convert_to: Type[TResultDataclass], strict: Literal[True] = True, extra: Extra | None = None, session: TSession | None = None) → TResultDataclass
get_by_filters(*, filters: IFilterSeq, strict: Literal[False], extra: Extra | None = None, session: TSession | None = None) → TResultORM | None
get_by_filters(*, filters: IFilterSeq, convert_to: Type[TResultDataclass], strict: Literal[False], extra: Extra | None = None, session: TSession | None = None) → TResultDataclass | None

Helper for @overload to raise when called.

get_by_pk(pk: TPrimaryKey, *, strict: Literal[True] = True, extra: Extra | None = None, session: TSession | None = None) → TResultORM
get_by_pk(pk: TPrimaryKey, *, convert_to: Type[TResultDataclass], strict: Literal[True] = True, extra: Extra | None = None, session: TSession | None = None) → TResultDataclass
get_by_pk(pk: TPrimaryKey, *, strict: Literal[False], extra: Extra | None = None, session: TSession | None = None) → TResultORM | None
get_by_pk(pk: TPrimaryKey, *, convert_to: Type[TResultDataclass], strict: Literal[False], extra: Extra | None = None, session: TSession | None = None) → TResultDataclass | None

Helper for @overload to raise when called.

is_soft_deletable: bool
multi_update(pks: Sequence[TPrimaryKey], *, values: Mapping[str, TFieldValue], extra: Extra | None = None, session: TSession | None = None) → None[source]

Update rows

Parameters:

  • pks (Sequence[TPrimaryKey]) – Primary keys of rows to update

  • values (Mapping[str, TFieldValue]) – Mapping with format {field_name:new_value}

  • extra (Extra | None, optional) – Extra params. Defaults to None

  • session (TSession | None) – Session to use for DB queries. Defaults to None. Currently supported for SQLAlchemy

pk_field_name: str
session_factory: AbstractContextManager | None
table_class: Type[TTable]
update(pk: TPrimaryKey, *, values: Mapping[str, TFieldValue], extra: Extra | None = None, session: TSession | None = None) → None[source]

Update row

Parameters:

  • pk (TPrimaryKey) – Primary key of row to update

  • values (Mapping[str, TFieldValue]) – Mapping with format {field_name:new_value}

  • extra (Extra | None, optional) – Extra params. Defaults to None

  • session (TSession | None) – Session to use for DB queries. Defaults to None. Currently supported for SQLAlchemy

dbrepos.core.exceptions module

exception dbrepos.core.exceptions.BaseRepoException[source]

Bases: Exception

dbrepos.core.types module

class dbrepos.core.types.Extra(for_update: bool = False, include_soft_deleted: bool = False, ordering: ~typing.Tuple[str, ...] = <factory>, select_related: ~typing.Tuple[str, ...] = <factory>)[source]

Bases: object

Parameters:

  • for_update (bool) – Lock row. Defaults to False

  • include_soft_deleted (bool) – Include soft deleted rows. Works only for soft deletable models. Defaults to False

  • ordering (Tuple[str]) – Result ordering. Defaults to empty tuple (meaning default ordering is applied)

  • select_related (Tuple[str]) – Columns to join from related tables. Defaults to empty tuple (meaning columns are joined)

for_update: bool = False
include_soft_deleted: bool = False
ordering: Tuple[str, ...]
class dbrepos.core.types.mode(*values)[source]

Bases: IntEnum

and_ = 0
or_ = 1
class dbrepos.core.types.operator(*values)[source]

Bases: IntEnum

eq = 0
ge = 4
gt = 3
in_ = 5
is_ = 6
le = 2
lt = 1

Module contents