Skip to content

Actions API

See the Actions guide for runnable examples of function actions, services, CRUD operations, and workflows. The data guide explains how DataS uses DataR for persistence.

WriteOperation 

WriteOperation = Literal[
    "create",
    "update",
    "patch",
    "delete",
    "bulk_create",
    "upsert",
    "bulk_upsert",
    "update_by_query",
    "delete_by_query",
    "bulk_update",
    "bulk_patch",
    "bulk_delete",
]

Action

Bases: Actionable[I, O]

Base for a class action with one typed input and output.

dependencies class-attribute instance-attribute 

dependencies: Sequence[DependsClass] = ()

__init__ 

__init__(*, dependencies: Sequence[DependsClass] | None = None) -> None

__init_subclass__ 

__init_subclass__() -> None

FunctionAction

Callable object so assignment to an action never binds self.

dependencies instance-attribute 

dependencies: Sequence[DependsClass] = dependencies

target instance-attribute 

target: Callable[P, Awaitable[R]] = function

function instance-attribute 

function: Callable[P, Awaitable[R]] = with_dependencies(function, dependencies)

__wrapped__ instance-attribute 

__wrapped__: Callable[P, Awaitable[R]] = function if source is None else source

dependency property 

dependency: DependsClass

__init__ 

__init__(
    function: Callable[P, Awaitable[R]],
    *,
    dependencies: Sequence[DependsClass] = (),
    source: Callable[P, Awaitable[R]] | None = None,
) -> None

__call__ async 

__call__(*args: P.args, **kwargs: P.kwargs) -> R

Service

Bases: ActionGroupable

__action_group__ class-attribute 

__action_group__: bool = True

dependencies class-attribute instance-attribute 

dependencies: Sequence[DependsClass] = ()

__init__ 

__init__(*, dependencies: Sequence[DependsClass] | None = None) -> None

__init_subclass__ 

__init_subclass__() -> None

DataS

Bases: Service, DataSable[E, R, C, U, P, CreateSetting]

CRUD actions using the entity or explicitly configured schemas.

operations class-attribute instance-attribute 

operations: frozenset[DataSOperation] = configuration.DATAS_OPERATIONS

chunk_size class-attribute instance-attribute 

chunk_size: Annotated[int, Ge(1)] = DEFAULT_ITERATION_CHUNK_SIZE

entity instance-attribute 

entity: type[E] = self.source(entity)

page_schema class-attribute instance-attribute 

page_schema: type[Page[R]] = Page

entity_type instance-attribute 

entity_type: type[E]

read_schema instance-attribute 

read_schema: type[R]

create_schema instance-attribute 

create_schema: type[C]

update_schema instance-attribute 

update_schema: type[U]

patch_schema instance-attribute 

patch_schema: type[P]

update class-attribute instance-attribute 

update = update

patch class-attribute instance-attribute 

patch = patch

delete class-attribute instance-attribute 

delete = delete

upsert class-attribute instance-attribute 

upsert = upsert

delete_by_query class-attribute instance-attribute 

delete_by_query = delete_by_query

update_by_query class-attribute instance-attribute 

update_by_query = update_by_query

bulk_delete class-attribute instance-attribute 

bulk_delete = bulk_delete

get class-attribute instance-attribute 

get = get

exists class-attribute instance-attribute 

exists = exists

count class-attribute instance-attribute 

count = count

list class-attribute instance-attribute 

list = list

find_first class-attribute instance-attribute 

find_first = find_first

find_exactly_one class-attribute instance-attribute 

find_exactly_one = find_exactly_one

find_all class-attribute instance-attribute 

find_all = find_all

stream class-attribute instance-attribute 

stream = stream

map class-attribute instance-attribute 

map = map

join class-attribute instance-attribute 

join = join

bulk_create class-attribute instance-attribute 

bulk_create = bulk_create

bulk_update class-attribute instance-attribute 

bulk_update = bulk_update

bulk_patch class-attribute instance-attribute 

bulk_patch = bulk_patch

bulk_upsert class-attribute instance-attribute 

bulk_upsert = bulk_upsert

__init__ 

__init__(
    entity: type[E] | None = None,
    *,
    read: SchemaSetting[R] = Undefined,
    create: type[C] | CreateSetting,
    update: SchemaSetting[U] = Undefined,
    patch: SchemaSetting[P] = Undefined,
    page: type[Page[R]] | None = None,
    dependencies: Sequence[DependsClass] | None = None,
    operations: frozenset[DataSOperation] | None = None,
    chunk_size: Annotated[int, Ge(1)] | None = None,
) -> None

__init__(
    entity: type[E] | None = None,
    *,
    read: SchemaSetting[R] = Undefined,
    create: type[C] | UndefinedType = Undefined,
    update: SchemaSetting[U] = Undefined,
    patch: SchemaSetting[P] = Undefined,
    page: type[Page[R]] | None = None,
    dependencies: Sequence[DependsClass] | None = None,
    operations: frozenset[DataSOperation] | None = None,
    chunk_size: Annotated[int, Ge(1)] | None = None,
) -> None

__init__(
    entity: type[E] | None = None,
    *,
    read: SchemaSetting[R] = Undefined,
    create: SchemaSetting[C] = Undefined,
    update: SchemaSetting[U] = Undefined,
    patch: SchemaSetting[P] = Undefined,
    page: type[Page[R]] | None = None,
    dependencies: Sequence[DependsClass] | None = None,
    operations: frozenset[DataSOperation] | None = None,
    chunk_size: Annotated[int, Ge(1)] | None = None,
) -> None

source 

source(entity: type[E] | None) -> type[E]

__init_subclass__ 

__init_subclass__() -> None

create async 

create(entity: E | C, datar: DataR = Depends()) -> R

DataSable

Bases: Protocol[E, R, C, U, P, CreateSetting]

entity instance-attribute 

entity: type[E]

read_schema instance-attribute 

read_schema: type[R]

create_schema instance-attribute 

create_schema: type[C]

update_schema instance-attribute 

update_schema: type[U]

patch_schema instance-attribute 

patch_schema: type[P]

page_schema instance-attribute 

page_schema: type[Page[R]]

operations instance-attribute 

operations: frozenset[DataSOperation]

dependencies instance-attribute 

dependencies: Sequence[DependsClass]

chunk_size instance-attribute 

chunk_size: Annotated[int, Ge(1)]

create async 

create(entity: E | C, datar: DataR = ...) -> R

patch async 

patch(
    identifier: IdentificatorType, data: E | P, /, *, datar: DataR = ...
) -> R

delete async 

delete(identifier: IdentificatorType, datar: DataR = ...) -> int

upsert async 

upsert(entity: E, datar: DataR = ...) -> R

bulk_create 

bulk_create(
    entities: AnyIterable[E | C], /, *, datar: DataR = ...
) -> AsyncCommonStreamable[R]

bulk_patch 

bulk_patch(
    entities: AnyIterable[tuple[IdentificatorType, E | P]],
    /,
    *,
    datar: DataR = ...,
) -> AsyncCommonStreamable[R]

bulk_upsert 

bulk_upsert(
    entities: AnyIterable[E], /, *, datar: DataR = ...
) -> AsyncCommonStreamable[R]

bulk_delete 

bulk_delete(
    identifiers: AnyIterable[IdentificatorType], /, *, datar: DataR = ...
) -> LazyAwait[int]

get async 

get[Statement](
    identifier: IdentificatorType | BaseQuery[E, Statement], datar: DataR = ...
) -> R | None

exists async 

exists[Statement](
    identifier: IdentificatorType | BaseQuery[E, Statement], datar: DataR = ...
) -> bool

count async 

count[Statement](
    query: BaseQuery[E, Statement] | None = ..., *, datar: DataR = ...
) -> int

list async 

list[Statement](
    query: BaseQuery[E, Statement] | None = ..., *, datar: DataR = ...
) -> Page[R]

find_first async 

find_first[Statement](
    query: BaseQuery[E, Statement], *, datar: DataR = ...
) -> R | None

find_exactly_one async 

find_exactly_one[Statement](
    query: BaseQuery[E, Statement] | FieldPredicate[E],
    *predicates: FieldPredicate[E],
    datar: DataR = ...,
) -> R

find_all async 

find_all[Statement](
    query: BaseQuery[E, Statement], *, datar: DataR = ...
) -> builtins.list[R]

map 

map[O: Schema, Statement](
    query: BaseQuery[E, Statement],
    mapper: MapperCallable[R, O],
    *,
    datar: DataR = ...,
) -> AsyncGenerator[O, None]

join 

join[Other: Entitieable, Left, Right, Key: Hashable, LeftOrm, RightOrm](
    query: BaseQuery[E, Left],
    other: BaseQuery[Other, Right],
    *,
    on: tuple[FieldRef[E, Key, LeftOrm], FieldRef[Other, Key, RightOrm]],
    datar: DataR = ...,
) -> AsyncGenerator[tuple[R, Other], None]

Page

Bases: BaseSchema

model_config class-attribute 

model_config: ConfigDict = ConfigDict(defer_build=True)

items instance-attribute 

items: list[R]

total instance-attribute 

total: Annotated[int, Field(ge=0)]

offset class-attribute instance-attribute 

offset: Annotated[int, Field(ge=0)] = 0

limit class-attribute instance-attribute 

limit: Annotated[int | None, Field(ge=0)] = None

Workflow

Bases: IteratorActionable[InputT, OutputT], Generic[InputT, OutputT, SourceT]

Yield each step's results and pass each element to the next step.

run instance-attribute 

run = first

branches instance-attribute 

branches: (
    tuple[
        Callable[[InputT], AsyncGenerator[SourceT, None]],
        Callable[[SourceT], BranchHandler[OutputT] | None],
    ]
    | None
) = None

__init__ 

__init__(first: Callable[[InputT], Awaitable[OutputT]]) -> None

__init__(first: Callable[[InputT], OutputT]) -> None

__init__(first: WorkflowAction[InputT, OutputT]) -> None

__init__(first: WorkflowAction[InputT, OutputT]) -> None

__call__ 

__call__(payload: InputT) -> AsyncGenerator[OutputT, None]

then 

then[N: Schema](
    next_action: WorkflowAction[OutputT, N],
) -> Workflow[InputT, N]

fork 

fork[A: Schema, B: Schema](
    left: Callable[[OutputT], Awaitable[A]],
    right: Callable[[OutputT], Awaitable[B]],
) -> Workflow[InputT, ForkResult[A, B]]

join 

join[N: Schema](
    next_action: WorkflowAction[OutputT, N],
) -> Workflow[InputT, N]

branch 

branch[T: Schema, N: Schema](
    kind: type[T], handler: WorkflowAction[T, N]
) -> Workflow[InputT, N, OutputT]

branch[T: Schema, N: Schema, Source: Schema | None](
    kind: type[T], handler: WorkflowAction[T, N]
) -> Workflow[InputT, OutputT | N, Source]

branch[T: Schema, N: Schema, Source: Schema | None](
    kind: type[T], handler: WorkflowAction[T, N]
) -> Workflow[InputT, N, OutputT] | Workflow[InputT, OutputT | N, Source]

ForkResult

Bases: Schema

left instance-attribute 

left: A

right instance-attribute 

right: B