First Component - PSP Docs

Folder Structure

Every component follows the same structure:

src/psp/components/notes/

notes/
├── __init__.py           # exports NoteModule
├── module.py             # component registration
│
├── domain/
│   ├── __init__.py
│   ├── models.py         # entities with invariants
│   ├── events.py         # domain events
│   └── errors.py         # domain-specific errors
│
├── application/
│   ├── __init__.py
│   ├── ports.py          # repository interfaces
│   └── use_cases/
│       ├── __init__.py
│       ├── create_note.py
│       ├── update_note.py
│       └── delete_note.py
│
└── adapters/
    ├── http/
    │   ├── __init__.py
    │   ├── router.py     # FastAPI routes
    │   └── schemas.py    # Pydantic models
    └── persistence/
        ├── __init__.py
        └── repo_memory.py  # in-memory implementation
                    

Step 1: Domain Models

Start with entities. They have no framework dependencies:

notes/domain/models.py

from dataclasses import dataclass, field
from datetime import datetime
from uuid import UUID

from .errors import EmptyContentError

@dataclass
class Note:
    id: UUID
    owner_id: UUID
    title: str
    content: str
    created_at: datetime
    updated_at: datetime

    def __post_init__(self):
        self._validate()

    def _validate(self):
        if not self.content.strip():
            raise EmptyContentError()

    def update(self, title: str, content: str, now: datetime):
        self.title = title
        self.content = content
        self.updated_at = now
        self._validate()
                    

notes/domain/errors.py

class NoteError(Exception):
    pass

class EmptyContentError(NoteError):
    def __init__(self):
        super().__init__("Note content cannot be empty")

class NoteNotFoundError(NoteError):
    def __init__(self, note_id: UUID):
        super().__init__(f"Note not found: {note_id}")
                    

Step 2: Domain Events

Define events for things that happen:

notes/domain/events.py

from dataclasses import dataclass
from datetime import datetime
from uuid import UUID

@dataclass(frozen=True)
class NoteCreated:
    note_id: UUID
    owner_id: UUID
    title: str
    created_at: datetime

@dataclass(frozen=True)
class NoteUpdated:
    note_id: UUID
    owner_id: UUID
    updated_at: datetime

@dataclass(frozen=True)
class NoteDeleted:
    note_id: UUID
    owner_id: UUID
                    

Step 3: Ports (Interfaces)

Define what the application layer needs:

notes/application/ports.py

from typing import Protocol
from uuid import UUID

from psp.platform.query import QuerySpec
from ..domain.models import Note

class NoteRepository(Protocol):
    def get(self, note_id: UUID) -> Note | None: ...
    def save(self, note: Note) -> None: ...
    def delete(self, note_id: UUID) -> None: ...
    def query(self, spec: QuerySpec) -> list[Note]: ...
                    

Step 4: Use Cases

Business operations with injected dependencies:

notes/application/use_cases/create_note.py

from dataclasses import dataclass
from uuid import UUID, uuid4

from psp.platform.clock import Clock
from psp.platform.eventbus import EventBus
from ..ports import NoteRepository
from ...domain.models import Note
from ...domain.events import NoteCreated

@dataclass
class CreateNoteInput:
    owner_id: UUID
    title: str
    content: str

class CreateNote:
    def __init__(
        self,
        repo: NoteRepository,
        clock: Clock,
        event_bus: EventBus
    ):
        self.repo = repo
        self.clock = clock
        self.event_bus = event_bus

    def execute(self, input: CreateNoteInput) -> Note:
        now = self.clock.now()

        note = Note(
            id=uuid4(),
            owner_id=input.owner_id,
            title=input.title,
            content=input.content,
            created_at=now,
            updated_at=now
        )

        self.repo.save(note)

        self.event_bus.publish(NoteCreated(
            note_id=note.id,
            owner_id=note.owner_id,
            title=note.title,
            created_at=now
        ))

        return note
                    

Step 5: Repository Implementation

Start with in-memory for development:

notes/adapters/persistence/repo_memory.py

from uuid import UUID

from psp.platform.query import QuerySpec, apply_filters
from ...domain.models import Note

class InMemoryNoteRepository:
    def __init__(self):
        self._notes: dict[UUID, Note] = {}

    def get(self, note_id: UUID) -> Note | None:
        return self._notes.get(note_id)

    def save(self, note: Note) -> None:
        self._notes[note.id] = note

    def delete(self, note_id: UUID) -> None:
        self._notes.pop(note_id, None)

    def query(self, spec: QuerySpec) -> list[Note]:
        notes = list(self._notes.values())
        return apply_filters(notes, spec)
                    

Step 6: HTTP Router

Thin layer that calls use cases:

notes/adapters/http/router.py

from fastapi import APIRouter, Depends, HTTPException
from uuid import UUID

from .schemas import CreateNoteRequest, NoteResponse
from .dependencies import get_create_note

router = APIRouter(prefix="/notes", tags=["notes"])

@router.post("/", response_model=NoteResponse)
def create_note(
    request: CreateNoteRequest,
    create_note: CreateNote = Depends(get_create_note)
):
    note = create_note.execute(CreateNoteInput(
        owner_id=request.owner_id,
        title=request.title,
        content=request.content
    ))
    return NoteResponse.from_domain(note)
                    

Step 7: Module Registration

Wire everything together:

notes/module.py

from psp.runtime.module import Module
from .adapters.http.router import router
from .adapters.persistence.repo_memory import InMemoryNoteRepository

class NoteModule(Module):
    def __init__(self):
        self.repo = InMemoryNoteRepository()

    @property
    def name(self) -> str:
        return "notes"

    @property
    def routers(self):
        return [router]

    @property
    def capabilities(self):
        return {"note_repo": self.repo}
                    

Step 8: Register in Main

Add to the application:

src/psp/main.py

from psp.components.notes import NoteModule

MODULES = [
    # ... existing modules
    NoteModule(),
]
                    