clean-architecture-python

name	clean-architecture-python
description	Hexagonal/Clean Architecture patterns for Python with ports-and-adapters, DDD, repository pattern, and dependency injection
triggers	["clean architecture","hexagonal","ports and adapters","domain driven design","DDD","repository pattern","dependency injection","SOLID"]

Clean Architecture for Python

Hexagonal architecture (ports-and-adapters) for Python services. Domain-driven design with clear layer boundaries.

Layer Structure

src/jarvis/
  domain/           # Entities, value objects, domain services (NO external deps)
    entities/
    value_objects/
    services/
    exceptions.py
  application/      # Use cases, ports (interfaces), DTOs
    use_cases/
    ports/           # Abstract interfaces (inbound + outbound)
    dtos/
  infrastructure/   # Adapters (implementations of ports)
    persistence/     # Database repos, ORM models
    external/        # API clients, message queues
    config/          # Settings, DI container
  api/              # HTTP/WebSocket adapters (FastAPI routes)
    routes/
    middleware/

Dependency Rule

Dependencies point INWARD only:

• api -> application -> domain
• infrastructure -> application -> domain
• domain has ZERO external imports

Port (Interface)

# application/ports/agent_repository.py
from abc import ABC, abstractmethod
from domain.entities.agent import Agent

class AgentRepository(ABC):
    @abstractmethod
    async def get_by_id(self, agent_id: str) -> Agent | None: ...

    @abstractmethod
    async def save(self, agent: Agent) -> Agent: ...

    @abstractmethod
    async def list_active(self) -> list[Agent]: ...

Adapter (Implementation)

# infrastructure/persistence/postgres_agent_repo.py
from application.ports.agent_repository import AgentRepository
from domain.entities.agent import Agent

class PostgresAgentRepository(AgentRepository):
    def __init__(self, session_factory):
        self._session_factory = session_factory

    async def get_by_id(self, agent_id: str) -> Agent | None:
        async with self._session_factory() as session:
            row = await session.get(AgentModel, agent_id)
            return row.to_domain() if row else None

Use Case

# application/use_cases/create_agent.py
from dataclasses import dataclass
from application.ports.agent_repository import AgentRepository
from domain.entities.agent import Agent

@dataclass
class CreateAgentUseCase:
    repo: AgentRepository  # Injected port, not concrete impl

    async def execute(self, name: str, agent_type: str) -> Agent:
        agent = Agent.create(name=name, agent_type=agent_type)
        return await self.repo.save(agent)

Dependency Injection

# infrastructure/config/container.py
from dependency_injector import containers, providers

class Container(containers.DeclarativeContainer):
    config = providers.Configuration()
    db_session = providers.Singleton(create_session_factory, config.database_url)
    agent_repo = providers.Factory(PostgresAgentRepository, session_factory=db_session)
    create_agent = providers.Factory(CreateAgentUseCase, repo=agent_repo)

Testing Each Layer

• Domain: Pure unit tests, no mocks needed (no external deps)
• Application: Mock the ports, test use case logic
• Infrastructure: Integration tests with real DB (Testcontainers)
• API: HTTP tests with TestClient, mock use cases