BaseService

BaseService orchestrates domain logic, coordinates transaction boundaries via _safe_commit(), manages soft-delete recovery lifecycles, and exposes comprehensive single and batch (_multi) lifecycle interception hooks.

Class Definition

from typing import Generic, TypeVar
from zcore import BaseService
from zcore.db.setup import Base

ModelType = TypeVar("ModelType", bound=Base)

class BaseService(
    Generic[ModelType],
    ReadServiceMixin[ModelType],
    WriteServiceMixin[ModelType],
    SearchServiceMixin[ModelType]
):
    def __init__(self, model: type[ModelType], repository: "BaseRepository"):
        ...

Properties

Read Methods

get

Fetches a single domain entity by dynamic filters and applies the post_get hook.

async def get(
    self, 
    *criterion: Any, 
    fields: list[Any] | None = None, 
    options: list[ExecutableOption] | None = None,
    **filters: Any
) -> ModelType: ...

get_by_ids

Fetches a sequence of domain entities by their identifiers, applying the post_get_multi hook.

async def get_by_ids(
    self, 
    ids: list[Any], 
    fields: list[Any] | None = None, 
    options: list[ExecutableOption] | None = None
) -> Sequence[ModelType]: ...

get_list

Fetches a paginated or complete listing of entities, applying the post_get_multi hook.

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]: ...

exist

Verifies the existence of a target entity matching criteria in the database.

async def exist(self, *criterion: Any, **filters: Any) -> bool: ...

count

Counts the volume of entities matching dynamic criteria.

async def count(self, *criterion: Any, **filters: Any) -> int: ...

Write, Mutation & Restoration Methods

create

Orchestrates the creation of an entity, merging pre_create data, calling on_create, executing post_create, and invoking _safe_commit().

async def create(self, schema: BaseModel, **extra_data: Any) -> ModelType: ...

create_multi

Orchestrates batch creation of multiple domain entities with pre_create_multi / post_create_multi lifecycle hooks and safe commit.

async def create_multi(
    self, 
    schemas: list[BaseModel], 
    refresh: bool = False
) -> Sequence[ModelType]: ...

update

Orchestrates modifications to an existing entity, merging pre_update data, calling on_update, executing post_update, and invoking _safe_commit().

async def update(
    self, 
    target: ModelType | Any, 
    schema: BaseModel, 
    partial: bool = False, 
    **extra_data: Any
) -> ModelType: ...

update_multi

Orchestrates batch modifications to multiple existing domain entities with pre_update_multi / post_update_multi hooks.

async def update_multi(
    self, 
    data: dict[ModelType | Any, BaseModel], 
    partial: bool = False, 
    refresh: bool = False
) -> Sequence[ModelType]: ...

delete

Orchestrates the deletion of a single entity, triggering pre_delete(target, force), calling on_delete(target, force), executing post_delete(model, force), and invoking _safe_commit().

async def delete(self, target: ModelType | Any, force: bool = False) -> ModelType: ...

delete_multi

Orchestrates batch deletions of multiple database records matching primary keys with pre_delete_multi(ids, force) / post_delete_multi(models, force) hooks.

async def delete_multi(self, ids: list[Any], force: bool = False) -> Sequence[ModelType]: ...

restore

Orchestrates the restoration of a soft-deleted domain entity, triggering pre_restore, calling on_restore, executing post_restore, and invoking _safe_commit().

async def restore(self, target: ModelType | Any) -> ModelType: ...

restore_multi

Orchestrates batch restorations of multiple soft-deleted records with pre_restore_multi / post_restore_multi hooks and safe commit.

async def restore_multi(self, ids: list[Any]) -> Sequence[ModelType]: ...

Search Methods

search

Triggers pre_search, executes on_search (with optional fields load_only and options), and processes results through post_search.

async def search(
    self, 
    search_in: SearchRequest, 
    pagination: Any = None,
    fields: list[Any] | None = None,
    options: list[ExecutableOption] | None = None,
) -> Sequence[ModelType] | PaginatedResult[ModelType]: ...

Core Execution Methods (on_*)

These methods directly delegate to the bound repository and can be overridden to customize persistence mechanics:

• async def on_create(self, schema: BaseModel, **extra_data: Any) -> ModelType
• async def on_create_multi(self, schemas: list[BaseModel], refresh: bool = False) -> Sequence[ModelType]
• async def on_update(self, target: ModelType | Any, schema: BaseModel, partial: bool = False, **extra_data: Any) -> ModelType | None
• async def on_update_multi(self, data: dict[ModelType | Any, BaseModel], partial: bool = False, refresh: bool = False) -> Sequence[ModelType]
• async def on_delete(self, target: ModelType | Any, force: bool = False) -> ModelType | None
• async def on_delete_multi(self, ids: list[Any], force: bool = False) -> Sequence[ModelType]
• async def on_restore(self, target: ModelType | Any) -> ModelType | None
• async def on_restore_multi(self, ids: list[Any]) -> Sequence[ModelType]
• async def on_search(self, search_in: SearchRequest, pagination: Any = None, fields: list[Any] | None = None, options: list[ExecutableOption] | None = None) -> Any

Lifecycle Hooks Reference

Override these hooks in your service to execute custom business logic before or after operations: