Dependencies
@action(dependencies=[Depends(check), ...]) runs dependencies in order
before the action body. An exception stops the call. Dependencies can use
the action's arguments by name and request other dependencies with Depends.
Set Service.dependencies to run checks before every public async action.
Pass dependencies=[...] to the Service constructor to override them for
one instance.
For a function decorated with @action, check.dependency is shorthand for
Depends(check). It also works for that function assigned to a Service,
as in checks.access.dependency below. For ordinary Service methods and
Action instances, use Depends(...).
Here, publishing and archiving both require access and content checks.
Publishing an empty document fails before publish runs.
"""Use actions and selected service methods as dependencies.
Service dependencies run before each public action. A function decorated
with @action exposes .dependency as a shortcut for Depends(action).
Exceptions stop the operation.
"""
import asyncio
from fastsprout.actions import Action, Service, action
from fastsprout.core import BaseSchema, Depends, Field
class Document(BaseSchema):
content: Field[str]
class CheckContent(Action[Document, Document]):
async def __call__(self, document: Document) -> Document:
if not document.content.strip():
raise ValueError("Document content is required")
print("Content checked:", document.content)
return document
@action
async def check_access(document: Document) -> None:
print("Access checked:", document.content)
class DocumentChecks(Service):
access = check_access
checks = DocumentChecks()
check_content = CheckContent()
class Documents(Service):
dependencies = (checks.access.dependency, Depends(check_content))
async def publish(self, document: Document) -> Document:
print("Published:", document.content)
return document
@action(dependencies=[Depends(checks.access), Depends(check_content)])
async def archive(document: Document) -> Document:
print("Archived:", document.content)
return document
async def main() -> None:
documents = Documents()
await documents.publish(Document(content="Invoice"))
await documents.publish(Document(content="Receipt"))
await archive(Document(content="Invoice"))
try:
await documents.publish(Document(content=""))
except ValueError as error:
print("Rejected:", error)
if __name__ == "__main__":
asyncio.run(main())
Run from the repository root: