Packages

Path Patterns

Example: Entity UserSession

from datetime import datetime
from uuid import UUID

from pydantic import BaseModel, SecretStr, field_serializer

class UserSession(BaseModel):
    id: UUID
    user_id: UUID
    created_at: datetime
    last_use_at: datetime
    secret: SecretStr

Application Layer (src/core/{context}/)

The Application layer contains use cases and orchestration logic: commands, queries, events, and ports. Everything in the bounded context folder except domain/.

Structure

src/core/{context}/
├── domain/         # Domain layer (see above)
├── commands/       # Commands and their Handlers (write operations)
├── queries/        # Queries and their Views (read operation definitions)
├── events/         # Domain Events
├── ports/          # Interfaces (Protocols) for dependency inversion
│   └── repo/       # Repository interfaces
└── shared/         # Shared code within the bounded context

Packages

Path Patterns

Example: Command with Handler

from typing import NamedTuple
from uuid import UUID

from cq import command_handler
from pydantic import BaseModel, SecretStr, field_serializer

from src.core.auth.domain.session import UserSession
from src.core.auth.ports.repo.user_permission import UserPermissionRepository
from src.core.auth.ports.repo.user_session import UserSessionRepository
from src.core.auth.ports.token_generator import TokenGenerator
from src.core.auth.shared.access_token import encode_access_token
from src.core.auth.shared.session_token import encode_session_token
from src.services.datetime.abc import DateTimeService
from src.services.hasher.abc import Hasher
from src.services.jwt.abc import JWTService
from src.services.uuid.abc import UUIDGenerator

class OpenUserSessionCommand(BaseModel):
    user_id: UUID

class UserTokens(BaseModel):
    access_token: SecretStr
    session_token: SecretStr

    @field_serializer("access_token", "session_token", when_used="json")
    def _dump_secret(self, value: SecretStr) -> str:
        return value.get_secret_value()

@command_handler
class OpenUserSessionHandler(NamedTuple):
    datetime: DateTimeService
    hasher: Hasher
    jwt: JWTService
    repo: UserSessionRepository
    token_generator: TokenGenerator
    user_permission_repo: UserPermissionRepository
    uuid: UUIDGenerator

    async def handle(self, command: OpenUserSessionCommand) -> UserTokens:
        user_id = command.user_id

        session_secret = self.token_generator.generate(128)
        session = self.new_session(user_id, session_secret)
        await self.repo.save(session)

        permissions = await self.user_permission_repo.get(user_id)
        access_token = encode_access_token(self.jwt, user_id, permissions)
        session_token = encode_session_token(session.id, session_secret)
        return UserTokens(
            access_token=SecretStr(access_token),
            session_token=SecretStr(session_token),
        )

    def new_session(self, user_id: UUID, session_secret: str) -> UserSession:
        now = self.datetime.utcnow()
        return UserSession(
            id=self.uuid.next(),
            user_id=user_id,
            created_at=now,
            last_use_at=now,
            secret=SecretStr(self.hasher.hash(session_secret)),
        )

Example: Port (Repository Interface)

from abc import abstractmethod
from typing import Protocol
from uuid import UUID

from src.core.auth.domain.session import UserSession

class UserSessionRepository(Protocol):
    @abstractmethod
    async def delete(self, session_id: UUID) -> None:
        raise NotImplementedError

    @abstractmethod
    async def get(self, session_id: UUID) -> UserSession | None:
        raise NotImplementedError

    @abstractmethod
    async def save(self, session: UserSession) -> None:
        raise NotImplementedError

Example: Query and View

from datetime import datetime
from uuid import UUID

from pydantic import BaseModel

class GetPrivateUserProfileQuery(BaseModel):
    user_id: UUID

class PrivateUserProfileView(BaseModel):
    id: UUID
    created_at: datetime
    first_name: str
    last_name: str

Shared Services (src/services/)

The Services layer defines abstract interfaces for common technical services used across the entire application. These are cross-cutting concerns that can be used by any layer.

Structure

src/services/{service_name}/
├── abc.py          # Abstract interface (Protocol)
└── {impl}.py       # Implementation

Path Patterns

Example: Abstract Service Hasher

from abc import abstractmethod
from typing import Protocol

class Hasher(Protocol):
    @abstractmethod
    def hash(self, value: str) -> str:
        raise NotImplementedError

    @abstractmethod
    def verify(self, value: str, hashed_value: str) -> bool:
        raise NotImplementedError

    def needs_rehash(self, hashed_value: str) -> bool:
        return False

Example: Implementation Argon2Hasher

from argon2 import PasswordHasher
from argon2.exceptions import InvalidHashError, VerificationError
from injection import injectable

from src.services.hasher.abc import Hasher

@injectable(on=Hasher)
class Argon2Hasher(Hasher):
    def __init__(self) -> None:
        self.__internal = PasswordHasher()

    def hash(self, value: str) -> str:
        return self.__internal.hash(value)

    def verify(self, value: str, hashed_value: str) -> bool:
        try:
            return self.__internal.verify(hashed_value, value)
        except (InvalidHashError, VerificationError):
            return False

    def needs_rehash(self, hashed_value: str) -> bool:
        return self.__internal.check_needs_rehash(hashed_value)

Infrastructure Layer (src/infra/)

The Infrastructure layer contains all concrete implementations: API, database, external integrations, etc.

Structure

src/infra/
├── adapters/           # Port implementations (repositories, services)
│   └── {context}/
│       └── repo/       # SQLAlchemy repositories
├── api/
│   ├── builder.py      # FastAPI configuration
│   ├── dependencies.py # FastAPI dependencies (auth, locale, etc.)
│   └── routes/         # Endpoints by domain
├── cli/
│   ├── builder.py      # Typer configuration
│   └── apps/           # CLI commands
├── db/                 # Database
│   ├── tables.py       # SQLAlchemy table definitions
│   └── migrations/     # Alembic migrations
├── integrations/       # Third-party integrations (Stripe, etc.)
│   └── {provider}/
│       └── commands/   # Integration-specific commands
└── query_handlers/     # Query handlers (DB read operations)

Packages

| Package | Role | Justification | |----------------------------------|---------