Todos Component

Task management with recurrence, reminders, and explainability

Todos Component Architecture
01

Overview

The Todos component is a full-featured task management system that demonstrates all PSP architectural patterns. It includes tasks, containers (lists/projects), tags, notes, and relationships between tasks.

src/psp/components/todos/ 
todos/
├── __init__.py           # exports TodosModule, TodosComponent
├── component.py          # composition root (framework-agnostic)
├── module.py             # catalogue registration metadata
│
├── domain/
│   ├── models/
│   │   ├── task.py       # Task entity with lifecycle
│   │   ├── container.py  # TaskContainer (list, project, area)
│   │   ├── tag.py        # Tag for categorization
│   │   ├── note.py       # TaskNote (append-only)
│   │   └── relation.py   # TaskLink (blocks, depends_on)
│   ├── events.py         # 16 domain events
│   ├── errors.py         # domain-specific errors
│   └── facets.py         # user preferences
│
├── application/
│   ├── dto.py            # input/output DTOs
│   ├── hooks.py          # hook dispatch helpers
│   ├── ports/
│   │   └── repos.py      # 5 repository protocols
│   └── use_cases/
│       ├── create_task.py
│       ├── complete_task.py  # handles recurrence
│       ├── explain.py        # explainability queries
│       └── ...               # 20+ use cases
│
└── adapters/
    ├── http/
    │   ├── router.py         # combines sub-routers
    │   ├── router_tasks.py   # task endpoints
    │   ├── router_containers.py
    │   ├── router_tags.py
    │   ├── router_notes.py
    │   ├── router_links.py
    │   ├── router_explain.py
    │   ├── dependencies.py   # DI wiring (400 lines)
    │   └── schemas.py        # Pydantic models
    └── persistence/
        ├── repo_memory.py    # in-memory implementations
        ├── repo_sqlite.py    # SQLite implementations
        ├── sqlite_db.py      # database connection & schema
        └── query_builder.py  # QuerySpec → SQL translation
02

Domain Entities

Task

The core entity representing a unit of work. Tasks are immutable—mutation methods return new instances.

todos/domain/models/task.py 
@dataclass(frozen=True)
class Task:
    id: UUID
    owner_id: UUID              # Person or Group
    container_id: UUID | None   # which list/project
    title: str
    description: str | None
    status: TaskStatus          # PENDING | IN_PROGRESS | COMPLETED
    priority: Priority          # LOW | MEDIUM | HIGH | URGENT
    due_at: datetime | None
    completed_at: datetime | None
    deleted_at: datetime | None  # soft-delete flag
    recurrence: RecurrenceRule | None
    tags: frozenset[UUID]

    def complete(self, now: datetime) -> Task:
        """Return new task with COMPLETED status."""
        return dataclasses.replace(
            self,
            status=TaskStatus.COMPLETED,
            completed_at=now
        )

    def reopen(self) -> Task:
        """Return new task back to PENDING."""
        ...

    def soft_delete(self, now: datetime) -> Task:
        """Return new task marked as deleted."""
        ...

TaskContainer

Organizes tasks into lists, projects, areas, or board columns. Implements the HierarchyNode protocol for tree structures.

todos/domain/models/container.py 
class ContainerType(Enum):
    LIST = "list"
    PROJECT = "project"
    AREA = "area"
    BOARD_COLUMN = "board_column"
    CONTEXT = "context"
    CUSTOM = "custom"

@dataclass(frozen=True)
class TaskContainer:
    id: UUID
    owner_id: UUID
    name: str
    container_type: ContainerType
    parent_id: UUID | None      # hierarchical nesting
    position: int               # sort order
    is_archived: bool

Tag, TaskNote, TaskLink

  • Tag — Cross-container labels with color and description
  • TaskNote — Append-only annotations (immutable after creation)
  • TaskLink — Directional relationships (BLOCKS, DEPENDS_ON, PARENT, RELATED)
03

Ports (Interfaces)

All persistence is abstracted through Protocol-based ports. This enables testability and swappable implementations.

todos/application/ports/repos.py 
class TaskRepository(Protocol):
    def get(self, task_id: UUID) -> Task | None:
        """Get task by ID, excluding soft-deleted."""
        ...

    def get_including_deleted(self, task_id: UUID) -> Task | None:
        """Get task by ID, including soft-deleted."""
        ...

    def save(self, task: Task) -> None: ...
    def query(self, spec: QuerySpec) -> QueryResult[Task]: ...
    def count(self, spec: QuerySpec) -> int: ...

class TaskContainerRepository(Protocol): ...
class TagRepository(Protocol): ...
class TaskNoteRepository(Protocol): ...
class TaskLinkRepository(Protocol): ...
Port Key Methods
TaskRepository get, save, query, count, get_including_deleted
TaskContainerRepository get, save, query, list_by_owner
TagRepository get, save, get_by_name, list_by_owner
TaskNoteRepository get, save, list_by_task, count_by_task
TaskLinkRepository get, save, delete, list_by_source, list_by_target
04

Use Cases

Use cases orchestrate domain logic with platform primitives. Each use case takes injected dependencies—no globals or service locators.

CreateTask

Demonstrates the full pattern: authorization, idempotency, hooks, persistence, and event publishing.

todos/application/use_cases/create_task.py 
class CreateTask:
    def __init__(
        self,
        task_repo: TaskRepository,
        clock: Clock,                    # Platform: time abstraction
        event_bus: EventBus,             # Platform: event publishing
        policy: Policy,                  # Platform: authorization
        hooks: HookDispatcher,           # Platform: extensibility
        idempotency: IdempotencyStore,   # Platform: deduplication
    ):
        self.task_repo = task_repo
        self.clock = clock
        ...

    def execute(self, input: CreateTaskInput) -> TaskOutput:
        # 1. Check idempotency
        if existing := self.idempotency.get(input.idempotency_key):
            return existing

        # 2. Authorize
        self.policy.require(input.actor_id, "create", input.owner_id, "task")

        # 3. Dispatch before-hook (can veto or patch)
        result = self.hooks.dispatch(BEFORE_CREATE_TASK, input)
        if result.vetoed:
            raise TaskCreationVetoedError(result.veto_reason)
        input = result.patched_data or input

        # 4. Create domain entity
        now = self.clock.now()
        task = Task(id=uuid4(), owner_id=input.owner_id, ...)

        # 5. Persist
        self.task_repo.save(task)
        self.idempotency.store(input.idempotency_key, task_to_output(task))

        # 6. Publish event
        self.event_bus.publish(TaskCreated(task_id=task.id, ...))

        # 7. Dispatch after-hook
        self.hooks.dispatch(AFTER_CREATE_TASK, task)

        return task_to_output(task)

CompleteTask with Recurrence

When completing a recurring task, the use case creates the next occurrence automatically:

todos/application/use_cases/complete_task.py (simplified) 
def execute(self, input: CompleteTaskInput) -> TaskOutput:
    task = self.task_repo.get(input.task_id)

    # Mark current instance complete
    completed_task = task.complete(self.clock.now())
    self.task_repo.save(completed_task)

    # If recurring, create next occurrence
    if task.recurrence:
        next_due = self.recurrence_engine.next_occurrence(
            task.recurrence,
            task.due_at
        )
        next_task = Task(
            id=uuid4(),
            owner_id=task.owner_id,
            title=task.title,
            due_at=next_due,
            recurrence=task.recurrence,
            ...
        )
        self.task_repo.save(next_task)
        self.event_bus.publish(RecurrenceTriggered(...))

    return task_to_output(completed_task)

Use Case Categories

Category Use Cases
Task CRUD CreateTask, UpdateTask, DeleteTask, GetTask, ListTasks
Task Lifecycle CompleteTask, ReopenTask, MoveTask
Recurrence SetRecurrence, ClearRecurrence
Reminders RequestReminder, CancelReminder, GetTaskReminders
Containers CreateContainer, ArchiveContainer, ListContainers
Tags CreateTag, UpdateTag, DeleteTag, ListTags
Notes AddTaskNote, ListTaskNotes
Links LinkTasks, UnlinkTasks, GetBlockingTasks
Explainability ExplainTaskState, ExplainTaskPriority, GetTaskHistory
05

Component Composition

The TodosComponent class is the framework-agnostic composition root. It wires up all use cases with their dependencies, making the component usable from any context (HTTP, CLI, tests, background workers).

todos/component.py 
class TodosComponent:
    """Framework-agnostic composition root."""

    def __init__(
        self,
        # Repositories
        task_repo: TaskRepository,
        container_repo: TaskContainerRepository,
        tag_repo: TagRepository,
        note_repo: TaskNoteRepository,
        link_repo: TaskLinkRepository,
        # Platform services
        clock: Clock,
        event_bus: EventBus,
        policy: Policy,
        hooks: HookDispatcher,
        idempotency: IdempotencyStore,
        recurrence_engine: RecurrenceEngine,
        reminder_scheduler: ReminderScheduler,
        activity_log: ActivityLogRepo,
    ) -> None:
        # Wire up all use cases internally
        self.create_task = CreateTask(task_repo, container_repo, clock, ...)
        self.complete_task = CompleteTask(task_repo, clock, event_bus, ...)
        self.list_tasks = ListTasks(task_repo)
        # ... all 30+ use cases

Factory Function

The create_todos_component() factory provides common configurations:

todos/component.py 
def create_todos_component(
    storage: Literal["memory", "sqlite"] = "memory",
    db_path: str = "data/todos.db",
    *,
    clock: Clock | None = None,
    event_bus: EventBus | None = None,
    # ... other overrides
) -> TodosComponent:
    """Factory for common configurations."""
    if storage == "sqlite":
        repos, _db = create_sqlite_repositories(db_path)
    else:
        repos = create_memory_repositories()

    return TodosComponent(
        task_repo=repos.tasks,
        clock=clock or SystemClock(),
        event_bus=event_bus or InMemoryEventBus(),
        ...
    )

# Usage
component = create_todos_component()  # in-memory
component = create_todos_component(storage="sqlite")  # production

HTTP Adapter

The HTTP layer (dependencies.py) wraps the component for FastAPI. It creates a single component instance and exposes use cases via Depends():

todos/adapters/http/dependencies.py 
# Create component once - this is the composition root
_component = create_todos_component(
    storage="sqlite" if _USE_SQLITE else "memory",
    db_path=_DB_PATH,
)

def get_component() -> TodosComponent:
    return _component

# Thin wrappers for FastAPI Depends()
def get_create_task(
    component: TodosComponent = Depends(get_component),
) -> CreateTask:
    return component.create_task  # No wiring here!

def get_complete_task(
    component: TodosComponent = Depends(get_component),
) -> CompleteTask:
    return component.complete_task
Component Architecture
%%{init: {"flowchart": {"useMaxWidth": false}}}%% graph TB FACTORY[create_todos_component] --> COMP[TodosComponent] COMP --> UC1[create_task] COMP --> UC2[complete_task] COMP --> UC3[list_tasks] COMP --> UCN[...] DEPS[dependencies.py] --> COMP HTTP[HTTP Router] --> DEPS COMP -.-> REPO[(Repositories)] COMP -.-> PLAT[Platform Services] style COMP fill:#f3e8ff,stroke:#9333ea style FACTORY fill:#d1fae5,stroke:#059669 style HTTP fill:#ecfeff,stroke:#0891b2
Design Benefits
  • Framework-agnostic — Component works without FastAPI (CLI, workers, tests)
  • Single composition root — All wiring in one place, not scattered in adapters
  • Testable — Inject mock dependencies directly into component
  • Clear boundary — TodosComponent is the component
06

Persistence Layer

The Todos component provides two repository implementations: in-memory for development/testing and SQLite for production persistence.

In-Memory Repositories

Simple dict-based storage for fast iteration during development:

todos/adapters/persistence/repo_memory.py 
class InMemoryTaskRepository:
    def __init__(self):
        self._tasks: dict[UUID, Task] = {}

    def get(self, task_id: UUID) -> Task | None:
        task = self._tasks.get(task_id)
        if task and task.deleted_at is None:
            return task
        return None

    def save(self, task: Task) -> None:
        self._tasks[task.id] = task

    def query(self, spec: QuerySpec) -> QueryResult[Task]:
        # Filter, sort, paginate in-memory
        ...

SQLite Repositories

Production-ready persistence with proper schema, indexing, and thread-safe connections.

todos/adapters/persistence/sqlite_db.py 
# Schema with proper indexes
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS tasks (
    id TEXT PRIMARY KEY,
    owner_id TEXT NOT NULL,
    container_id TEXT NOT NULL,
    title TEXT NOT NULL,
    description TEXT,
    status TEXT NOT NULL DEFAULT 'pending',
    priority TEXT NOT NULL DEFAULT 'medium',
    due_at TEXT,
    completed_at TEXT,
    deleted_at TEXT,
    recurrence TEXT,      -- JSON for RecurrenceState
    tags TEXT,            -- JSON array of UUIDs
    extensions TEXT,      -- JSON for TaskExtensions
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_tasks_owner ON tasks(owner_id);
CREATE INDEX IF NOT EXISTS idx_tasks_container ON tasks(container_id);
CREATE INDEX IF NOT EXISTS idx_tasks_status ON tasks(status);
CREATE INDEX IF NOT EXISTS idx_tasks_deleted ON tasks(deleted_at);
"""

class SQLiteDatabase:
    """Thread-local connections for thread safety."""

    def __init__(self, db_path: str):
        self._db_path = Path(db_path)
        self._local = threading.local()

    def initialize(self) -> None:
        """Create tables on first use."""
        conn = self._get_connection()
        conn.executescript(SCHEMA_SQL)
        conn.commit()
todos/adapters/persistence/repo_sqlite.py 
class SQLiteTaskRepository:
    def __init__(self, db: SQLiteDatabase):
        self._db = db

    def get(self, task_id: UUID) -> Task | None:
        row = self._db.fetchone(
            "SELECT * FROM tasks WHERE id = ? AND deleted_at IS NULL",
            (str(task_id),)
        )
        return self._row_to_task(row) if row else None

    def save(self, task: Task) -> None:
        self._db.execute(
            """INSERT OR REPLACE INTO tasks
               (id, owner_id, container_id, title, ...)
               VALUES (?, ?, ?, ?, ...)""",
            (str(task.id), str(task.owner_id), ...)
        )

    def query(self, spec: QuerySpec) -> QueryResult[Task]:
        builder = SQLiteQueryBuilder("tasks")
        builder.add_condition("deleted_at IS NULL")
        sql, params = builder.build_select(spec)
        rows = self._db.fetchall(sql, params)
        return QueryResult(items=[self._row_to_task(r) for r in rows], ...)

QuerySpec to SQL Translation

The SQLiteQueryBuilder translates platform QuerySpec into SQL:

todos/adapters/persistence/query_builder.py 
class SQLiteQueryBuilder:
    """Translates QuerySpec to SQLite SQL statements."""

    def build_select(self, spec: QuerySpec) -> tuple[str, list]:
        self._apply_filters(spec.filters)  # EQ, NEQ, IN, LT, GT...
        self._apply_text_search(spec.text) # title/description LIKE

        sql = f"SELECT * FROM {self._table}"
        if self._conditions:
            sql += " WHERE " + " AND ".join(self._conditions)
        sql += self._build_order_by(spec.sort)
        sql += self._build_pagination(spec.page)

        return sql, self._params

    def _apply_filters(self, filters: list[Filter]) -> None:
        for f in filters:
            if f.op == FilterOp.EQ:
                self._conditions.append(f"{f.field} = ?")
            elif f.op == FilterOp.IN:
                placeholders = ",".join("?" * len(f.value))
                self._conditions.append(f"{f.field} IN ({placeholders})")
            # ... NEQ, LT, LTE, GT, GTE, CONTAINS, EXISTS

Data Mapping

Complex domain fields are serialized to JSON for SQLite storage:

Domain Field SQLite Type Serialization
UUID TEXT str(uuid)
datetime TEXT ISO8601 format
Enum TEXT enum.value
list[UUID] (tags) TEXT JSON array
RecurrenceState TEXT JSON object
TaskExtensions TEXT JSON object

Switching Implementations

Swap implementations in dependencies.py:

todos/adapters/http/dependencies.py 
# Development: in-memory
_task_repo = InMemoryTaskRepository()

# Production: SQLite
_db = SQLiteDatabase("data/todos.db")
_db.initialize()
_task_repo = SQLiteTaskRepository(_db)
_container_repo = SQLiteTaskContainerRepository(_db)
_tag_repo = SQLiteTagRepository(_db)
_note_repo = SQLiteTaskNoteRepository(_db)
_link_repo = SQLiteTaskLinkRepository(_db)
07

HTTP Routes

Routers are thin layers that parse requests, call use cases, and return responses. All routes are prefixed with /v1/todos.

todos/adapters/http/router_tasks.py (example) 
@router.post("/", response_model=TaskResponse)
def create_task(
    request: CreateTaskRequest,
    create_task: CreateTask = Depends(get_create_task)
):
    output = create_task.execute(CreateTaskInput(
        owner_id=request.owner_id,
        title=request.title,
        description=request.description,
        ...
    ))
    return task_output_to_response(output)

@router.post("/{task_id}/complete", response_model=TaskResponse)
def complete_task(
    task_id: UUID,
    complete_task: CompleteTask = Depends(get_complete_task)
):
    output = complete_task.execute(CompleteTaskInput(task_id=task_id))
    return task_output_to_response(output)

API Endpoints

Method Path Description
POST /tasks Create task
GET /tasks List tasks (with filters)
GET /tasks/{id} Get single task
PATCH /tasks/{id} Update task
POST /tasks/{id}/complete Mark complete
POST /tasks/{id}/reopen Reopen completed
POST /tasks/{id}/delete Soft delete
POST /containers Create container
POST /tags Create tag
POST /links Link tasks
GET /explain/task/{id}/state Explain current state
08

Platform Integrations

The Todos component uses all PSP platform primitives:

Primitive Usage
Clock Abstract time for testability. All timestamps via clock.now()
EventBus Publish TaskCreated, TaskCompleted, etc. for subscribers
HookDispatcher Before/after hooks for create, complete, update. Budget uses this to veto task completion.
Policy Authorization checks on mutations
IdempotencyStore Prevent duplicate creates on retries
QuerySpec Filtering, sorting, pagination on list endpoints
09

Module Registration

The TodosModule class registers the component with the application host:

todos/module.py 
from psp.runtime.module import Module

class TodosModule(Module):
    name = "todos"
    version = "0.1.0"

    @property
    def capabilities(self) -> list[str]:
        return ["commands", "queries", "events", "hooks"]

    def router(self, ctx: RuntimeContext) -> APIRouter:
        return create_router()

    def describe(self) -> ComponentSpec:
        """Return full introspection metadata for the catalogue."""
        return ComponentSpec(
            name=self.name,
            ports=self._get_defined_ports(),
            schemas=self._get_schemas(),
            hooks=self._get_hooks(),
            events=self._get_events(),
            stories=self._get_stories(),
            examples=self._get_examples(),
        )

The module is registered in main.py:

src/psp/main.py 
from psp.components.todos import TodosModule
from psp.components.budget import BudgetModule
from psp.components.catalogue import CatalogueModule

host = ApplicationHost(modules=[
    TodosModule(),
    BudgetModule(),
    CatalogueModule(),
])

app = FastAPI()
host.mount_routers(app, prefix="/v1")
10

Domain Events

The Todos component publishes 16 domain events. Other components can subscribe to react to task lifecycle changes.

todos/domain/events.py 
# Task events
@dataclass(frozen=True)
class TaskCreated:
    task_id: UUID
    owner_id: UUID
    title: str
    created_at: datetime

@dataclass(frozen=True)
class TaskCompleted:
    task_id: UUID
    owner_id: UUID
    completed_at: datetime

@dataclass(frozen=True)
class RecurrenceTriggered:
    original_task_id: UUID
    new_task_id: UUID
    next_due: datetime

# Container events
TaskUpdated, TaskReopened, TaskDeleted, TaskMoved, TaskOverdue
ContainerCreated, ContainerArchived, ContainerMoved

# Tag/annotation events
TagCreated, TagAddedToTask, TagRemovedFromTask
NoteAddedToTask, TasksLinked, TasksUnlinked
Key Patterns
  • Immutable entities — Domain models are frozen dataclasses; mutations return new instances
  • Soft deletes — Deleted tasks have deleted_at set; queries exclude them by default
  • Protocol-based ports — All persistence abstracted for testability
  • Constructor injection — Use cases receive all dependencies in __init__
  • Thin routers — HTTP handlers only parse → call → return
  • Event-driven — All mutations publish domain events