Architecture overview โ
Layer structure โ
nene2-python follows Clean Architecture. Dependencies flow from the outside in.
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ HTTP Handler (FastAPI router) โ
โ parse request โ call use-case โ response โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ UseCase โ
โ Business logic โ no HTTP or DB knowledge โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ RepositoryInterface (ABC) โ
โ Contract for the operations the domain โ
โ needs โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโค
โ ConcreteRepository โ
โ SQLAlchemy / InMemory implementations โ
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโLayer responsibilities โ
HTTP Handler โ
- Single responsibility: parse the request, call a UseCase, return a response
- Uses Pydantic
BaseModelfor request body validation (HTTP boundary only) - Contains zero domain logic
- Exposed via a
make_xxx_router()factory function
python
@router.post("", status_code=201)
async def create_note(body: CreateNoteBody) -> JSONResponse:
note = create_use_case.execute(CreateNoteInput(title=body.title, body=body.body))
return JSONResponse({"id": note.id, "title": note.title, "body": note.body}, status_code=201)UseCase โ
- Single responsibility: implement one business rule
- One method:
execute(input_: XxxInput) -> XxxOutput - No
import fastapi, noimport sqlalchemy - Does not call other UseCases
- Testable with
InMemoryRepositoryalone
RepositoryInterface โ
- Defined as an ABC โ the UseCase depends only on the interface
- Same interface is implemented by InMemory and SQLAlchemy versions
find_all,find_by_id,save,update,delete,count
ConcreteRepository โ
- SQLAlchemy Core (no ORM) with parameterised queries
- Queries are executed via
SqlAlchemyQueryExecutor - Table schema: the example app uses a central
src/example/schema.py; for new projects, defineensure_schema()in each domain'ssqlalchemy_repository.pyand call each fromcreate_app()
Middleware stack โ
Requests pass through each middleware from outermost to innermost:
BearerTokenMiddleware Authentication (Bearer Token)
ApiKeyAuthMiddleware Authentication (API Key)
CORSMiddleware CORS
ThrottleMiddleware Rate limiting (fixed window)
RequestSizeLimitMiddleware Payload size enforcement
RequestLoggingMiddleware Structured request logging (structlog)
RequestIdMiddleware X-Request-ID generation / propagation
SecurityHeadersMiddleware Security response headers
ErrorHandlerMiddleware Exceptions โ RFC 9457 Problem DetailsDependency injection โ
FastAPI's Depends is used at the HTTP boundary only. UseCases and repositories are wired via constructor injection in app.py.
python
# app.py โ wiring
note_repo = SqlAlchemyNoteRepository(executor)
app.include_router(make_note_router(
list_use_case=ListNotesUseCase(note_repo),
create_use_case=CreateNoteUseCase(note_repo),
...
))Domain package layout โ
src/example/<domain>/
__init__.py
entity.py โ @dataclass(frozen=True, slots=True)
repository.py โ ABC + InMemory implementation
exceptions.py โ XxxNotFoundException + ExceptionHandler
use_case.py โ 5 UseCases + Input/Output DTOs
handler.py โ FastAPI router factory
sqlalchemy_repository.py โ SQL backend