7. Connect actions in a workflow
One operation can have several stages. Here, preparing a task must run the shared title check and normalization before a formatting step uses the result. Compose those stages as a reusable workflow.
The example reuses NewTask, TaskDraft, and the checked preparation action
from the earlier steps. The new operation accepts a TaskDraft and returns
another TaskDraft with an uppercase title.
Save this program as docs_src/tutorial/workflows.py.
"""Connect the checked preparation action to a typed formatting step."""
import asyncio
from docs_src.quickstart import NewTask, TaskDraft
from docs_src.tutorial.shared_checks import prepare_task
from fastsprout.actions import Workflow, action
@action
async def uppercase_title(draft: TaskDraft) -> TaskDraft:
return TaskDraft(title=draft.title.upper())
workflow = Workflow(prepare_task).then(uppercase_title)
async def main() -> None:
async for draft in workflow(NewTask(title=" Write documentation ")):
print(draft.title)
if __name__ == "__main__":
asyncio.run(main())
Run it from your tutorial directory:
Result
Follow the result through the steps
Workflow(prepare_task) starts with an action that takes NewTask and returns
TaskDraft. .then(uppercase_title) connects that result to the next action's
input. Their types remain available to the type checker.
Calling the workflow returns an async generator. Iteration starts execution, runs the shared check, and passes each result to the next step. This example emits one draft; the same interface can handle steps that emit no result or multiple results.
A workflow packages the composition as one callable operation. As the process grows, it can include other workflows, concurrent forks, and typed branches. Continue with the workflow guide for those patterns, or return to the Data guide for entity-service schemas, bulk operations, and joins.