ManzarIMalik Python Backend Architecture
Production-ready patterns for Python backend systems following Clean Architecture, Domain-Driven Design, and modern distributed system patterns.
Quick Reference
| Need to... | See Reference |
|---|---|
| Structure a new project | project-structure.md |
| Follow code standards | code-style.md |
| Implement DI, Repository, UoW | architecture-patterns.md |
| Design entities and value objects | domain-driven-design.md |
| Build FastAPI endpoints | api-design.md |
| Set up SQLAlchemy and queries | database-patterns.md |
| Implement caching | caching.md |
| Handle events and sagas | event-driven.md |
| Build microservices | microservices.md |
| Secure the application | security.md |
| Add logging, metrics, tracing | observability.md |
| Write tests | testing.md |
| Deploy to production | deployment.md |
Core Principles
Architecture Layers
┌─────────────────────────────────────────────┐
│ Presentation Layer │ ← API routes, schemas, middleware
├─────────────────────────────────────────────┤
│ Application Layer │ ← Use cases, DTOs, interfaces
├─────────────────────────────────────────────┤
│ Domain Layer │ ← Entities, value objects, events
├─────────────────────────────────────────────┤
│ Infrastructure Layer │ ← DB, cache, messaging, external APIs
└─────────────────────────────────────────────┘
Dependency Rule: Dependencies point inward. Domain has no external dependencies.
Key Principles
- Dependency Injection - Depend on abstractions, not implementations
- Single Responsibility - One reason to change per class/function
- Explicit over Implicit - No hidden dependencies or magic
- Async by Default - Use async/await for all I/O operations
- Type Hints Everywhere - All public APIs fully typed
- Immutability - Prefer immutable data structures (especially value objects)
- Early Return - Reduce nesting with guard clauses
Architecture Decision Flowchart
flowchart TD
A[New Feature] --> B{Complex domain logic?}
B -->|Yes| C[Use Domain-Driven Design]
B -->|No| D{Multiple data sources?}
C --> E[Define Entities & Value Objects]
E --> F[Create Repository Interfaces]
F --> G[Implement Use Cases]
D -->|Yes| H[Use Repository Pattern]
D -->|No| I{Transaction across aggregates?}
H --> G
I -->|Yes| J[Use Unit of Work]
I -->|No| K[Direct repository calls]
J --> G
K --> G
G --> L{Cross-service operation?}
L -->|Yes| M[Use Saga Pattern]
L -->|No| N[Single transaction]
M --> O[Publish Domain Events]
N --> O
O --> P{Needs real-time updates?}
P -->|Yes| Q[Event-Driven Architecture]
P -->|No| R[Request-Response]
Common Patterns Summary
Domain Layer
- Entities: Objects with identity and lifecycle. See domain-driven-design.md
- Value Objects: Immutable objects compared by value. See domain-driven-design.md
- Domain Events: Capture what happened. See domain-driven-design.md
- Aggregates: Consistency boundaries. See domain-driven-design.md
Application Layer
- Use Cases: Single operation orchestrators
- DTOs: Data transfer between layers
- Interfaces: Abstract ports for infrastructure
Infrastructure Layer
- Repositories: Data access abstraction. See architecture-patterns.md
- Unit of Work: Transaction management. See architecture-patterns.md
- Event Bus: Publish/subscribe messaging. See event-driven.md
Presentation Layer
- Routers: HTTP endpoint handlers. See api-design.md
- Schemas: Request/response validation. See api-design.md
- Dependencies: DI factory functions. See api-design.md
Error Handling Strategy
# Domain errors (business logic violations)
class DomainError(Exception): pass
class EntityNotFoundError(DomainError): pass
class BusinessRuleViolationError(DomainError): pass
# Infrastructure errors (external system failures)
class InfrastructureError(Exception): pass
class DatabaseConnectionError(InfrastructureError): pass
class ExternalAPIError(InfrastructureError): pass
# Map to HTTP in presentation layer
@router.get("/{id}")
async def get_item(id: UUID, use_case: GetItemUseCase = Depends(...)):
try:
return await use_case.execute(id)
except EntityNotFoundError:
raise HTTPException(status_code=404, detail="Not found")
except BusinessRuleViolationError as e:
raise HTTPException(status_code=400, detail=str(e))
except InfrastructureError:
raise HTTPException(status_code=503, detail="Service unavailable")
See code-style.md for full exception hierarchy.
Quick Start Patterns
Basic Repository + Use Case
# 1. Define interface (application layer)
class UserRepository(ABC):
@abstractmethod
async def get_by_id(self, id: UUID) -> User | None: pass
@abstractmethod
async def save(self, user: User) -> User: pass
# 2. Implement (infrastructure layer)
class PostgresUserRepository(UserRepository):
def __init__(self, session: AsyncSession):
self._session = session
async def get_by_id(self, id: UUID) -> User | None:
result = await self._session.execute(
select(UserModel).where(UserModel.id == id)
)
model = result.scalar_one_or_none()
return self._to_entity(model) if model else None
# 3. Use case (application layer)
@dataclass
class GetUserUseCase:
user_repo: UserRepository
async def execute(self, user_id: UUID) -> UserDTO:
user = await self.user_repo.get_by_id(user_id)
if not user:
raise EntityNotFoundError("User", str(user_id))
return UserDTO.from_entity(user)
# 4. Route (presentation layer)
@router.get("/{user_id}")
async def get_user(
user_id: UUID,
use_case: GetUserUseCase = Depends(get_user_use_case)
):
return await use_case.execute(user_id)
Adding Caching
@dataclass
class GetUserUseCase:
user_repo: UserRepository
cache: CacheService
async def execute(self, user_id: UUID) -> UserDTO:
# Try cache
cached = await self.cache.get(f"user:{user_id}")
if cached:
return UserDTO.parse_raw(cached)
# Fetch from DB
user = await self.user_repo.get_by_id(user_id)
if not user:
raise EntityNotFoundError("User", str(user_id))
# Cache result
dto = UserDTO.from_entity(user)
await self.cache.set(f"user:{user_id}", dto.json(), ttl=300)
return dto
See caching.md for advanced patterns.
Publishing Events
@dataclass
class CreateOrderUseCase:
order_repo: OrderRepository
event_bus: EventBus
async def execute(self, command: CreateOrderCommand) -> OrderDTO:
order = Order.create(user_id=command.user_id, items=command.items)
saved = await self.order_repo.add(order)
# Publish event after successful save
await self.event_bus.publish(OrderCreatedEvent.from_entity(saved))
return OrderDTO.from_entity(saved)
See event-driven.md for event handlers and saga patterns.
Production Readiness
Before deploying, verify:
- Observability: Logging, metrics, tracing configured. See observability.md
- Security: Auth, input validation, rate limiting. See security.md
- Testing: Unit, integration, E2E tests passing. See testing.md
- Resilience: Circuit breakers, retries, graceful shutdown. See microservices.md
- Deployment: Docker, K8s, health checks configured. See deployment.md