Actions, services, and workflows

fastsprout.actions defines application operations with typed inputs, results, and dependencies. Call an action directly, group related actions in a service, or compose them into a workflow. The same operation can be used from an HTTP handler, CLI command, or event subscriber. This keeps application checks and behavior in one callable operation as the number of entry points grows. See Features for the problems the layers address together. For a guided introduction, start with one action, then add checks, services, persistence, and a workflow through the seven tutorial steps.

Install FastSprout 0.2.0 from PyPI with the standard extra, or follow the tutorial setup:

pip install 'fastsprout[standard]==0.2.0'

Use	Start with
One async operation	`@action` or `Action[Input, Output]`
Related operations with shared dependencies	`Service` or an `@action` class
CRUD operations for an entity	`DataS`
A pipeline of typed operations	`Workflow`

Function actions

Decorate an async function with @action. Its parameter and result types remain available to the type checker. Calling it runs its dependencies and returns an awaitable result:

"""Action — function with Pydantic objects."""

import asyncio
from typing import assert_type

from fastsprout.actions import action
from fastsprout.core import BaseSchema, Field


class HeroIn(BaseSchema):
    name: Field[str]


class HeroOut(BaseSchema):
    name: Field[str]


@action
async def create_hero(payload: HeroIn) -> HeroOut:
    return HeroOut(name=payload.name)


async def main() -> None:
    hero = await create_hero(HeroIn(name="Batman"))

    assert_type(hero, HeroOut)
    print("HeroOut:", hero)


if __name__ == "__main__":
    asyncio.run(main())

Save it as action_example.py and run it in the environment where you installed FastSprout:

python action_example.py

For a class with a single operation, inherit from Action[Input, Output] and implement async __call__. Its input and output types must inherit Schema: use BaseSchema, or inherit Schema in your dataclass or custom class. See the runnable class action, dataclass action, and custom object action examples.

Services and dependency injection

Use an @action class to group public async methods. Depends resolves collaborators, including dependencies declared as attributes:

"""Action service — standard."""

import asyncio
from typing import assert_type

from fastsprout.actions import action
from fastsprout.core import BaseSchema, Depends, Field


class Document(BaseSchema):
    content: Field[str]


class SignedDocument(BaseSchema):
    content: Field[bytes]
    signature: Field[str]


class DocumentRenderer:
    async def render(self, document: Document) -> bytes:
        return document.content.encode()


class Signer:
    async def sign(self, content: bytes) -> str:
        return f"signed:{len(content)}"


@action
class DocumentService:
    renderer: DocumentRenderer = Depends()
    signer: Signer = Depends()

    async def render(self, document: Document) -> bytes:
        return await self.renderer.render(document)

    async def sign(self, document: Document) -> SignedDocument:
        content = await self.render(document)

        return SignedDocument(
            content=content,
            signature=await self.signer.sign(content),
        )


async def main() -> None:
    documents = DocumentService()
    document = Document(content="FastSprout")

    content = await documents.render(document)
    assert_type(content, bytes)

    signed = await documents.sign(document)
    assert_type(signed, SignedDocument)
    print(signed)


if __name__ == "__main__":
    asyncio.run(main())

Inheriting from Service provides the same action-group behavior without applying the decorator. See the Service example.

Use dependencies=[Depends(check)] on an action or service to run checks before its body. An exception stops the call. Dependencies can use the call's arguments by name and request other dependencies. The dependencies guide covers shared checks and overrides.

Dependency resolution also works inside an invocation context. The runnable dependency context and injection context examples show how to share context with nested calls.

Entity services with DataS

DataS(Entity) is an application service for an entity. It replaces repeated CRUD handlers with typed operations backed by DataR. DataR routes those operations to persistence and shares their backend context. The data guide explains their responsibilities. The SQLite tutorial introduces entity operations before adding a shared transaction and queries.

DataS uses the entity as its default read and write schema. Explicit read, create, update, and patch schemas let the service expose separate contracts for each operation. The operations setting controls which operations are available.

Operations	Purpose
`get`, `exists`, `count`	Look up an entity or inspect matching records
`list`, `find_first`, `find_exactly_one`, `find_all`, `stream`	Read pages, individual results, or streams
`create`, `update`, `patch`, `delete`, `upsert`	Write individual records
`bulk_create`, `bulk_update`, `bulk_patch`, `bulk_delete`, `bulk_upsert`	Process batches
`update_by_query`, `delete_by_query`	Write records selected by a query
`map`, `join`	Compose reads

Page[ReadSchema] carries items, total, offset, and limit for paginated reads. Streaming operations keep iteration lazy. Backend lifetime and transaction boundaries belong to the data context used by the service.

Start with the runnable CRUD service, then see custom schemas. The Actions API documents the service methods and DataSable protocol.

Workflows

Create a Workflow from its first operation, then append steps with .then(...). Each step receives the preceding step's result. Calling the workflow returns an async generator; execution starts when you iterate it:

"""Compose three typed actions in a linear workflow.
Each result becomes the next action input, including the rendered document."""

import asyncio

from fastsprout.actions import Workflow, action
from fastsprout.core import BaseSchema


class Document(BaseSchema):
    content: str


class RenderedDocument(BaseSchema):
    content: bytes


@action
async def parse(document: Document) -> Document:
    return Document(content=document.content.strip())


@action
async def validate(document: Document) -> Document:
    if not document.content:
        raise ValueError("Document is empty")

    return document


@action
async def render(document: Document) -> RenderedDocument:
    return RenderedDocument(content=document.content.encode())


workflow = Workflow(parse).then(validate).then(render)


async def main() -> None:
    async for result in workflow(Document(content="FastSprout")):
        print(result)


if __name__ == "__main__":
    asyncio.run(main())

Workflow inputs and outputs are schemas. A step can return one schema, None, a synchronous or asynchronous iterable of schemas, or an awaitable of a supported result. None emits no result. Each emitted item enters the next step separately, so a workflow may yield zero, one, or many outputs. Iteration releases resources when it completes, fails, or is closed. Close the async generator when you stop consuming it early, for example with contextlib.aclosing.

Composition	Behavior
`.then(next_action)`	Pass each result to the next operation
`.fork(left, right)`	Run two async branches concurrently and emit `ForkResult(left=..., right=...)`
`.join(next_action)`	Continue with an operation that accepts the combined result
`.branch(SchemaType, handler)`	Route by result type; the first matching branch wins

An unmatched typed branch raises LookupError. Exceptions from a step propagate to the caller. See the runnable workflow examples for linear composition, then browse the example catalog for forks, branching, streaming, and shared dependencies.

See Actions API for the public types and signatures.