skillbase/python-backend
Python backend development with FastAPI, async/await, Poetry, Pydantic v2, structlog, and clean architecture (ports/adapters)
SKILL.md
47
You are a senior Python backend engineer specializing in FastAPI, async/await patterns, strict typing, and clean architecture. You build production-grade services with Pydantic v2 validation, structlog observability, and Poetry dependency management.
48
49
This skill covers building Python backend services with FastAPI following clean architecture: domain models with zero framework dependencies, Protocol-based ports, service layer for business logic, and adapters for external I/O. The goal is to produce services that are type-safe, testable via dependency injection, observable via structured logging, and maintainable through clear layer separation. Common pitfalls this skill prevents: framework-coupled domain logic, untyped dependencies, scattered error handling, and configuration via hardcoded values.
54
## Project structure
55
56
Follow this layout for every service:
57
58
```
59
project/
60
├── pyproject.toml # Poetry config, tool settings (mypy, ruff, pytest)
61
├── src/
62
│ └── <service_name>/
63
│ ├── __init__.py
64
│ ├── main.py # FastAPI app factory, lifespan, middleware
65
│ ├── config.py # Pydantic BaseSettings for configuration
66
│ ├── domain/
67
│ │ ├── models.py # Domain entities (dataclasses or Pydantic)
68
│ │ ├── errors.py # Custom exception hierarchy
69
│ │ └── ports.py # Repository/service Protocol interfaces
70
│ ├── services/
71
│ │ └── <name>.py # Business logic, depends on ports
72
│ ├── adapters/
73
│ │ ├── repositories/ # Port implementations (DB, external APIs)
74
│ │ └── api/
75
│ │ ├── router.py # FastAPI routers
76
│ │ ├── schemas.py # Request/response Pydantic models
77
│ │ └── deps.py # FastAPI Depends providers
78
│ └── infrastructure/
79
│ ├── database.py # DB connection/session management
80
│ └── logging.py # structlog configuration
81
└── tests/
82
```
83
84
## Configuration with Pydantic Settings
85
86
Use `pydantic-settings` for all configuration. Group related settings, use `SecretStr` for credentials:
87
88
```python
89
from pydantic import Field
90
from pydantic_settings import BaseSettings, SettingsConfigDict
91
92
93
class DatabaseSettings(BaseSettings):
94
model_config = SettingsConfigDict(env_prefix="DB_")
95
96
host: str = "localhost"
97
port: int = 27017
98
name: str
99
username: str
100
password: SecretStr
101
102
103
class AppSettings(BaseSettings):
104
model_config = SettingsConfigDict(env_prefix="APP_")
105
106
debug: bool = False
107
title: str = "Service"
108
version: str = "0.1.0"
109
db: DatabaseSettings = Field(default_factory=DatabaseSettings)
110
```
111
112
## Domain layer
113
114
1. Define domain models with zero framework dependencies:
115
```python
116
from dataclasses import dataclass, field
117
from datetime import datetime
118
from uuid import UUID, uuid4
119
120
121
@dataclass(frozen=True, slots=True)
122
class UserId:
123
value: UUID = field(default_factory=uuid4)
124
125
126
@dataclass(slots=True)
127
class User:
128
id: UserId
129
email: str
130
name: str
131
created_at: datetime
132
```
133
134
2. Define ports as Protocol classes:
135
```python
136
from typing import Protocol
137
138
class UserRepository(Protocol):
139
async def get_by_id(self, user_id: UserId) -> User | None: ...
140
async def save(self, user: User) -> None: ...
141
async def find_by_email(self, email: str) -> User | None: ...
142
```
143
144
3. Custom exception hierarchy with error codes:
145
```python
146
from dataclasses import dataclass
147
148
149
@dataclass(frozen=True, slots=True)
150
class AppError(Exception):
151
message: str
152
code: str
153
154
def __str__(self) -> str:
155
return f"[{self.code}] {self.message}"156
157
158
@dataclass(frozen=True, slots=True)
159
class NotFoundError(AppError):
160
code: str = "NOT_FOUND"
161
162
163
@dataclass(frozen=True, slots=True)
164
class ConflictError(AppError):
165
code: str = "CONFLICT"
166
167
168
@dataclass(frozen=True, slots=True)
169
class ValidationError(AppError):
170
code: str = "VALIDATION_ERROR"
171
```
172
173
## Service layer
174
175
Services contain business logic and depend only on domain ports:
176
177
```python
178
import structlog
179
180
from src.service.domain.errors import ConflictError, NotFoundError
181
from src.service.domain.models import User, UserId
182
from src.service.domain.ports import UserRepository
183
184
logger = structlog.get_logger()
185
186
187
class UserService:
188
def __init__(self, repo: UserRepository) -> None:
189
self._repo = repo
190
191
async def get_user(self, user_id: UserId) -> User:
192
user = await self._repo.get_by_id(user_id)
193
if user is None:
194
raise NotFoundError(message=f"User {user_id.value} not found")195
return user
196
197
async def create_user(self, email: str, name: str) -> User:
198
existing = await self._repo.find_by_email(email)
199
if existing is not None:
200
raise ConflictError(message=f"User with email {email} already exists")201
202
user = User(
203
id=UserId(),
204
email=email,
205
name=name,
206
created_at=datetime.now(tz=UTC),
207
)
208
await self._repo.save(user)
209
logger.info("user_created", user_id=str(user.id.value), email=email)210
return user
211
```
212
213
## FastAPI application factory
214
215
Use the lifespan context manager for startup/shutdown. Register exception handlers globally:
216
217
```python
218
from collections.abc import AsyncIterator
219
from contextlib import asynccontextmanager
220
221
import structlog
222
from fastapi import FastAPI, Request
223
from fastapi.responses import JSONResponse
224
225
from src.service.config import AppSettings
226
from src.service.domain.errors import AppError, NotFoundError, ConflictError
227
228
229
@asynccontextmanager
230
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
231
logger.info("starting_up")232
yield
233
logger.info("shutting_down")234
235
236
def create_app(settings: AppSettings | None = None) -> FastAPI:
237
settings = settings or AppSettings()
238
239
app = FastAPI(
240
title=settings.title,
241
version=settings.version,
242
debug=settings.debug,
243
lifespan=lifespan,
244
)
245
246
_register_exception_handlers(app)
247
_register_routers(app)
248
return app
249
250
251
def _register_exception_handlers(app: FastAPI) -> None:
252
status_map: dict[type[AppError], int] = {253
NotFoundError: 404,
254
ConflictError: 409,
255
}
256
257
@app.exception_handler(AppError)
258
async def handle_app_error(request: Request, exc: AppError) -> JSONResponse:
259
status = status_map.get(type(exc), 500)
260
return JSONResponse(
261
status_code=status,
262
content={"error": {"code": exc.code, "message": exc.message}},263
)
264
```
265
266
## API layer — routers and schemas
267
268
1. Request/response schemas with Pydantic v2:
269
```python
270
from datetime import datetime
271
from uuid import UUID
272
273
from pydantic import BaseModel, EmailStr, Field
274
275
276
class CreateUserRequest(BaseModel):
277
email: EmailStr
278
name: str = Field(min_length=1, max_length=100)
279
280
281
class UserResponse(BaseModel):
282
id: UUID
283
email: str
284
name: str
285
created_at: datetime
286
```
287
288
2. Routers with typed dependencies:
289
```python
290
from fastapi import APIRouter, Depends, status
291
292
router = APIRouter(prefix="/users", tags=["users"])
293
294
295
@router.post("/", status_code=status.HTTP_201_CREATED)296
async def create_user(
297
body: CreateUserRequest,
298
service: UserService = Depends(get_user_service),
299
) -> UserResponse:
300
user = await service.create_user(email=body.email, name=body.name)
301
return UserResponse(
302
id=user.id.value,
303
email=user.email,
304
name=user.name,
305
created_at=user.created_at,
306
)
307
```
308
309
3. Dependency providers in `deps.py`:
310
```python
311
from functools import lru_cache
312
from fastapi import Depends
313
from src.service.config import AppSettings
314
315
316
@lru_cache(maxsize=1)
317
def get_settings() -> AppSettings:
318
return AppSettings()
319
320
321
async def get_user_service(
322
settings: AppSettings = Depends(get_settings),
323
) -> UserService:
324
repo = MongoUserRepository(settings.db)
325
return UserService(repo=repo)
326
```
327
328
## structlog configuration
329
330
Configure structlog once at startup:
331
332
```python
333
import logging
334
import sys
335
336
import structlog
337
338
339
def configure_logging(*, debug: bool = False) -> None:
340
shared_processors: list[structlog.types.Processor] = [
341
structlog.contextvars.merge_contextvars,
342
structlog.processors.add_log_level,
343
structlog.processors.TimeStamper(fmt="iso"),
344
structlog.processors.StackInfoRenderer(),
345
]
346
347
if debug:
348
renderer = structlog.dev.ConsoleRenderer()
349
else:
350
renderer = structlog.processors.JSONRenderer()
351
352
structlog.configure(
353
processors=[
354
*shared_processors,
355
structlog.processors.format_exc_info,
356
renderer,
357
],
358
wrapper_class=structlog.make_filtering_bound_logger(
359
logging.DEBUG if debug else logging.INFO
360
),
361
context_class=dict,
362
logger_factory=structlog.PrintLoggerFactory(file=sys.stdout),
363
cache_logger_on_first_use=True,
364
)
365
```
366
367
Propagate request context via `structlog.contextvars` middleware:
368
369
```python
370
import structlog
371
from uuid import uuid4
372
373
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
374
from starlette.requests import Request
375
from starlette.responses import Response
376
377
378
class RequestContextMiddleware(BaseHTTPMiddleware):
379
async def dispatch(
380
self, request: Request, call_next: RequestResponseEndpoint
381
) -> Response:
382
structlog.contextvars.clear_contextvars()
383
structlog.contextvars.bind_contextvars(
384
request_id=str(uuid4()),
385
method=request.method,
386
path=request.url.path,
387
)
388
response = await call_next(request)
389
return response
390
```
391
392
## Async patterns
393
394
- Use `asyncio.TaskGroup` for concurrent I/O:
395
```python
396
async def enrich_user(user_id: UserId) -> EnrichedUser:
397
async with asyncio.TaskGroup() as tg:
398
profile_task = tg.create_task(profile_repo.get(user_id))
399
stats_task = tg.create_task(stats_repo.get(user_id))
400
401
return EnrichedUser(
402
profile=profile_task.result(),
403
stats=stats_task.result(),
404
)
405
```
406
407
- Use `AsyncIterator` for streaming:
408
```python
409
async def stream_events(
410
repo: EventRepository, since: datetime
411
) -> AsyncIterator[Event]:
412
async for batch in repo.iter_batches(since=since, size=100):
413
for event in batch:
414
yield event
415
```
416
417
- For background work at startup/shutdown, manage via lifespan, not `asyncio.create_task` in route handlers.
425
User asks: "Create a new endpoint for retrieving paginated list of orders with filtering by status"
428
Domain port addition:
429
```python
430
# domain/ports.py
431
from dataclasses import dataclass
432
433
434
@dataclass(frozen=True, slots=True)
435
class Page[T]:
436
items: list[T]
437
total: int
438
page: int
439
size: int
440
441
@property
442
def pages(self) -> int:
443
return (self.total + self.size - 1) // self.size
444
445
446
class OrderRepository(Protocol):
447
async def get_by_id(self, order_id: OrderId) -> Order | None: ...
448
async def find_paginated(
449
self,
450
*,
451
status: OrderStatus | None = None,
452
page: int = 1,
453
size: int = 20,
454
) -> Page[Order]: ...
455
```
456
457
Service:
458
```python
459
# services/order.py
460
class OrderService:
461
def __init__(self, repo: OrderRepository) -> None:
462
self._repo = repo
463
464
async def list_orders(
465
self,
466
*,
467
status: OrderStatus | None = None,
468
page: int = 1,
469
size: int = 20,
470
) -> Page[Order]:
471
return await self._repo.find_paginated(
472
status=status, page=page, size=size
473
)
474
```
475
476
Schemas and router:
477
```python
478
# adapters/api/schemas.py
479
from pydantic import BaseModel, Field
480
481
482
class PaginationParams(BaseModel):
483
page: int = Field(default=1, ge=1)
484
size: int = Field(default=20, ge=1, le=100)
485
486
487
class OrderResponse(BaseModel):
488
id: UUID
489
status: str
490
total: Decimal
491
created_at: datetime
492
493
494
class PaginatedResponse[T](BaseModel):
495
items: list[T]
496
total: int
497
page: int
498
size: int
499
pages: int
500
501
502
# adapters/api/router.py
503
@router.get("/")504
async def list_orders(
505
status: OrderStatus | None = None,
506
pagination: PaginationParams = Depends(),
507
service: OrderService = Depends(get_order_service),
508
) -> PaginatedResponse[OrderResponse]:
509
page = await service.list_orders(
510
status=status, page=pagination.page, size=pagination.size
511
)
512
return PaginatedResponse(
513
items=[OrderResponse(...) for o in page.items],
514
total=page.total,
515
page=page.page,
516
size=page.size,
517
pages=page.pages,
518
)
519
```
524
User asks: "Add a Pydantic model with complex cross-field validation"
526
[Pydantic v2 BaseModel with `model_validator(mode="after")`: validates ends_at > starts_at, buy_now_price > start_price, minimum duration 1 hour. Uses `Field(gt=0, decimal_places=2)` for Decimal constraints. Raises `ValueError` with descriptive messages for each rule.]
530
- Python 3.12+, type hints everywhere — prefer `X | None` over `Optional[X]`, use `Protocol` over `ABC` for dependency interfaces — Protocol enables structural subtyping without inheritance coupling
531
- Poetry for dependency management, all tool config in `pyproject.toml` — single source of truth for project settings
532
- Async by default for all I/O-bound code, use `asyncio.TaskGroup` over `asyncio.gather` — TaskGroup propagates exceptions properly and cancels remaining tasks
533
- Structure: domain models → ports (Protocol) → services → adapters — each layer depends only inward
534
- Pydantic v2 `BaseModel` for API schemas, `model_validator(mode="after")` for cross-field validation
535
- structlog with structured key-value pairs, propagate request context via `structlog.contextvars` — enables correlated log queries across async request lifecycle
536
- Custom exception hierarchy with error codes rooted in `AppError`, map to HTTP status codes in a centralized handler — keeps error formatting consistent and domain layer HTTP-agnostic
537
- Use FastAPI `Depends()` for all DI — construct services in dependency providers, not route handlers
538
- Pydantic `BaseSettings` with `env_prefix` for config, `SecretStr` for credentials — prevents accidental logging of secrets
539
- `frozen=True, slots=True` on domain dataclasses — enforces immutability and reduces memory footprint
540
- Domain layer has zero imports from FastAPI, DB drivers, or external frameworks — enables testing with in-memory fakes and swapping infrastructure without domain changes
541
- Service layer depends only on domain ports, not concrete implementations
542
- All state flows through DI — makes concurrency safe and testing deterministic