4. Store and retrieve tasks
The feature needs to save a task and find it by ID. DataS(Task) supplies
those entity operations, so you can call create and get without writing
their implementations.
The TaskService from the previous step groups application checks and
formatting. This entity service stores and retrieves records. An application
operation can compose both when it needs to prepare a title and persist it.
Enable SQL support
Install the SQL extra and SQLite driver from PyPI in your active tutorial environment:
Create and get a task
The example setup supplies a Task entity with id, title, and completed
fields, plus a fresh in-memory database. The service is DataS(Task).
Save the shared setup under SQLite setup and the Task entity below as
docs_src/data/common.py, then save this program as docs_src/data/quickstart.py.
"""Create and retrieve a task using DataS's existing CRUD operations."""
import asyncio
from docs_src.data.common import Task, database
from fastsprout.actions import DataS
tasks = DataS(Task)
async def main() -> None:
async with database():
created = await tasks.create(Task(id=1, title="Write documentation"))
print("Created:", created.id, created.title)
found = await tasks.get(created.id)
if found is not None:
print("Found:", found.id, found.title)
print("Missing:", await tasks.get(999))
if __name__ == "__main__":
asyncio.run(main())
Run it from your tutorial directory:
Result
What DataS handles
create writes the entity and returns its read representation. get looks up
its ID and returns the entity, or None when it does not exist. This example
uses the entity as its default input and output schema.
DataS also supplies update, patch, delete, count, list, and bulk operations.
You can later choose separate input and output schemas and extend
DataS[Task] with application methods. See the Data guide
when you need those options.
SQLite setup and the Task entity
The examples share this file, docs_src/data/common.py:
"""SQLite setup shared by the tutorial's persistence examples."""
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlmodel import Field as SQLField
from fastsprout.core import Field, ReadField, inject_depends
from fastsprout.data import DataR, provide_datar
from fastsprout.data.backend.sql import SQLEntity, SQLSignpost
from fastsprout.data.backend.sql.entity import SQL_ENTITY_REGISTRY
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
store = SQLSignpost(async_sessionmaker(engine, expire_on_commit=False))
class Task(SQLEntity[int]):
__signpost__ = store
id: ReadField[int] = SQLField(primary_key=True)
title: Field[str]
completed: Field[bool] = Field(default=False)
@asynccontextmanager
async def database() -> AsyncIterator[None]:
inject_depends.dependency_providers[DataR] = provide_datar
try:
async with engine.begin() as connection:
await connection.run_sync(SQL_ENTITY_REGISTRY.metadata.create_all)
yield
finally:
await engine.dispose()
SQLSignpost connects the entity to its session factory. database() prepares
the tables and registers a dependency provider that gives each top-level
service call a managed data context. It disposes the engine when the example
ends. The next step makes the context boundary explicit so several calls can
share a transaction.
In this setup, successful service calls settle independently. That is useful for a single operation. The next step shows what happens when two writes must succeed or fail together.