BaseRepository
Comprehensive API reference for the BaseRepository class, query scoping, bulk operations, soft-delete lifecycles, and restoration methods.
BaseRepository provides standard asynchronous database access, combining Read, Write, Bulk Mutation, Soft-Delete Restoration, and Search capabilities. It decouples business logic from SQLAlchemy 2.0 sessions and enforces context security through model-level query scoping.
Class Definition
from typing import Generic, TypeVar
from sqlalchemy.ext.asyncio import AsyncSession
from zcore import BaseRepository
from zcore.db.setup import Base
ModelType = TypeVar("ModelType", bound=Base)
class BaseRepository(
Generic[ModelType],
ReadRepositoryMixin[ModelType],
WriteRepositoryMixin[ModelType],
SearchRepositoryMixin[ModelType]
):
def __init__(self, model: type[ModelType], db: AsyncSession):
...Properties
Prop
Type
Query Scoping & Soft Delete (_get_base_query)
BaseRepository constructs its initial SELECT statements via _get_base_query().
If your SQLAlchemy model declares a classmethod named scope_query (or inherits from SoftDeleteMixin), _get_base_query() automatically passes the query through this hook before applying any route-level filters:
# Internal mechanics in BaseRepository
def _get_base_query(self) -> Select:
query = select(self.model)
scoper = getattr(self.model, "scope_query", None)
if scoper:
return scoper(query)
return query
def _supports_soft_delete(self) -> bool:
"""Determine whether the underlying model implements SoftDeleteMixin."""
return hasattr(self.model, "deleted_at") and hasattr(self.model, "soft_delete")Model Scoping Hooks:
# Inheriting SoftDeleteMixin automatically filters out deleted records
class Task(Base, SoftDeleteMixin):
__tablename__ = "tasks"
...All read, listing, count, search, and existence queries automatically apply row-level scoping and soft-deletion filters.
Read Methods
get
Fetches a single model instance matching dynamic criteria.
async def get(
self,
*criterion: Any,
fields: list[Any] | None = None,
options: list[ExecutableOption] | None = None,
**filters: Any
) -> ModelType | None: ...Prop
Type
get_by_ids
Fetches a sequence of records matching a list of primary keys in a single query.
async def get_by_ids(
self,
ids: list[Any],
fields: list[Any] | None = None,
options: list[ExecutableOption] | None = None
) -> Sequence[ModelType]: ...Prop
Type
get_list
Fetches a paginated or complete listing of records.
async def get_list(
self,
*criterion: Any,
pagination: Any = None,
fields: list[Any] | None = None,
options: list[ExecutableOption] | None = None,
**filters: Any
) -> Sequence[ModelType] | PaginatedResult[ModelType]: ...Prop
Type
exist
Checks if records matching the filters exist in the database (automatically applies scope_query).
async def exist(self, *criterion: Any, **filters: Any) -> bool: ...Prop
Type
count
Counts records matching dynamic criteria.
async def count(self, *criterion: Any, **filters: Any) -> int: ...Prop
Type
Write, Bulk Mutation & Restoration Methods
create
Creates and persists a single record from a Pydantic schema.
async def create(self, schema: BaseModel, **extra_data: Any) -> ModelType: ...Prop
Type
create_multi
Creates multiple database records with dialect-aware fallback for returning support.
async def create_multi(
self,
schemas: list[BaseModel],
refresh: bool = False
) -> Sequence[ModelType]: ...Prop
Type
update
Updates an existing database record from a model instance or primary key.
async def update(
self,
target: ModelType | Any,
schema: BaseModel,
partial: bool = False,
**extra_data: Any
) -> ModelType | None: ...Prop
Type
update_multi
Bulk updates multiple database records using DBAPI executemany.
async def update_multi(
self,
data: dict[ModelType | Any, BaseModel],
partial: bool = False,
refresh: bool = False
) -> Sequence[ModelType]: ...Prop
Type
delete
Deletes a single record by its model instance or primary key identifier. Automatically applies soft-deletion if supported by the model unless force=True.
async def delete(self, target: ModelType | Any, force: bool = False) -> ModelType | None: ...Prop
Type
delete_multi
Deletes multiple records matching the provided list of primary keys. Automatically executes an atomic soft-delete update (deleted_at = now()) unless force=True.
async def delete_multi(self, ids: list[Any], force: bool = False) -> Sequence[ModelType]: ...Prop
Type
restore
Restores a previously soft-deleted record by primary key or model instance.
async def restore(self, target: ModelType | Any) -> ModelType | None: ...Prop
Type
restore_multi
Restores multiple soft-deleted records using an atomic batch UPDATE ... WHERE deleted_at IS NOT NULL statement with dialect-aware RETURNING optimization.
async def restore_multi(self, ids: list[Any]) -> Sequence[ModelType]: ...Prop
Type
Search Methods
search
Dynamically parses and executes a structured SearchRequest with automatic security checks, inverted operators, logical negation, column projection, and relation eager-loading.
async def search(
self,
search_in: SearchRequest,
pagination: Any = None,
fields: list[Any] | None = None,
options: list[ExecutableOption] | None = None,
) -> Sequence[ModelType] | PaginatedResult[ModelType]: ...Prop
Type
CLI & Structured Logging
Deep dive into ZCore's cascading server runner, introspection-based environment scaffolding, and unified structlog observability pipeline.
SearchEngine
API reference for the dynamic search engine, query builders, filter operators, inverted comparisons, and configurable relation depth.