Pattern di progettazione Python per un'architettura pulita
Costruisci applicazioni Python mantenibili con i pattern di progettazione SOLID
Clean Architecture ha rivoluzionato il modo in cui gli sviluppatori costruiscono applicazioni scalabili e mantenibili, enfatizzando la separazione delle preoccupazioni e la gestione delle dipendenze.
In Python, questi principi si combinano con la natura dinamica del linguaggio per creare sistemi flessibili e testabili che evolvono con i requisiti aziendali senza diventare debito tecnico.

Comprendere la Clean Architecture in Python
La Clean Architecture, introdotta da Robert C. Martin (Uncle Bob), organizza il software in strati concentrici in cui le dipendenze si dirigono verso l’interno verso la logica aziendale centrale. Questo modello architettonico garantisce che le regole critiche dell’applicazione rimangano indipendenti dai framework, dai database e dai servizi esterni.
La Filosofia Fondamentale
Il principio fondamentale è semplice ma potente: la logica aziendale non deve dipendere dall’infrastruttura. Le entità del dominio, i casi d’uso e le regole aziendali devono funzionare indipendentemente dal fatto che si utilizzi PostgreSQL o MongoDB, FastAPI o Flask, AWS o Azure.
In Python, questa filosofia si adatta perfettamente alla “duck typing” e alla programmazione orientata ai protocolli del linguaggio, permettendo una separazione pulita senza la cerimonia richiesta nei linguaggi tipizzati staticamente.
I Quattro Strati della Clean Architecture
Strato delle Entità (Dominio): Oggetti aziendali puri con regole aziendali valide in tutto l’azienda. Questi sono POJO (Plain Old Python Objects) senza dipendenze esterne.
Strato dei Casi d’Uso (Applicazione): Regole aziendali specifiche dell’applicazione che orchestrano il flusso di dati tra entità e servizi esterni.
Strato degli Adattatori dell’Interfaccia: Converte i dati tra il formato più conveniente per i casi d’uso e le entità e il formato richiesto dagli enti esterni.
Strato dei Framework e Driver: Tutti i dettagli esterni come database, framework web e API esterne.
Principi SOLID in Python
I principi SOLID formano la base dell’architettura pulita. Esploriamo come ciascun principio si manifesta in Python. Per un’overview completa dei pattern di progettazione in Python, vedi la Guida ai Pattern di Progettazione in Python.
Principio della Singola Responsabilità (SRP)
Ogni classe dovrebbe avere un solo motivo per cambiare:
# Cattivo: Multiple responsibilities
class UserManager:
def create_user(self, user_data):
# Create user
pass
def send_welcome_email(self, user):
# Send email
pass
def log_creation(self, user):
# Log to file
pass
# Buono: Responsabilità separate
class UserService:
def __init__(self, repository, email_service, logger):
self.repository = repository
self.email_service = email_service
self.logger = logger
def create_user(self, user_data):
user = User(**user_data)
self.repository.save(user)
self.email_service.send_welcome(user)
self.logger.info(f"Utente creato: {user.id}")
return user
Principio Aperto-Chiuso (OCP)
Le entità del software dovrebbero essere aperte per l’estensione ma chiuse per la modifica:
from abc import ABC, abstractmethod
from typing import Protocol
# Utilizzando Protocol (Python 3.8+)
class PaymentProcessor(Protocol):
def process_payment(self, amount: float) -> bool:
...
class CreditCardProcessor:
def process_payment(self, amount: float) -> bool:
# Logica per carta di credito
return True
class PayPalProcessor:
def process_payment(self, amount: float) -> bool:
# Logica per PayPal
return True
# Estendibile facilmente senza modificare il codice esistente
class CryptoProcessor:
def process_payment(self, amount: float) -> bool:
# Logica per criptovalute
return True
Principio di Sostituzione di Liskov (LSP)
Gli oggetti dovrebbero essere sostituibili con i loro sottotipi senza rompere il programma:
from abc import ABC, abstractmethod
class DataStore(ABC):
@abstractmethod
def save(self, key: str, value: str) -> None:
pass
@abstractmethod
def get(self, key: str) -> str:
pass
class PostgreSQLStore(DataStore):
def save(self, key: str, value: str) -> None:
# Implementazione PostgreSQL
pass
def get(self, key: str) -> str:
# Implementazione PostgreSQL
return ""
class RedisStore(DataStore):
def save(self, key: str, value: str) -> None:
# Implementazione Redis
pass
def get(self, key: str) -> str:
# Implementazione Redis
return ""
# Entrambi possono essere utilizzati in modo intercambiabile
def process_data(store: DataStore, key: str, value: str):
store.save(key, value)
return store.get(key)
Principio di Segregazione delle Interfacce (ISP)
I clienti non dovrebbero essere costretti a dipendere da interfacce che non utilizzano:
# Cattivo: Interfaccia grassa
class Worker(ABC):
@abstractmethod
def work(self): pass
@abstractmethod
def eat(self): pass
@abstractmethod
def sleep(self): pass
# Buono: Interfacce segregate
class Workable(Protocol):
def work(self) -> None: ...
class Eatable(Protocol):
def eat(self) -> None: ...
class Human:
def work(self) -> None:
print("Lavorando")
def eat(self) -> None:
print("Mangiando")
class Robot:
def work(self) -> None:
print("Lavorando")
# Nessun metodo eat necessario
Principio di Inversione delle Dipendenze (DIP)
I moduli di alto livello non dovrebbero dipendere da moduli di basso livello. Entrambi dovrebbero dipendere da astrazioni:
from typing import Protocol
# Astrazione
class EmailSender(Protocol):
def send(self, to: str, subject: str, body: str) -> None:
...
# Modulo di basso livello
class SMTPEmailSender:
def send(self, to: str, subject: str, body: str) -> None:
# Implementazione SMTP
pass
# Modulo di alto livello che dipende dall'astrazione
class UserRegistrationService:
def __init__(self, email_sender: EmailSender):
self.email_sender = email_sender
def register(self, email: str, name: str):
# Logica di registrazione
self.email_sender.send(
to=email,
subject="Benvenuto!",
body=f"Ciao {name}"
)
Pattern del Repository: Astrazione dell’Accesso ai Dati
Il Pattern del Repository fornisce un’interfaccia simile a una collezione per accedere agli oggetti del dominio, nascondendo i dettagli del deposito dei dati.
Implementazione Base del Repository
from abc import ABC, abstractmethod
from typing import List, Optional
from dataclasses import dataclass
from uuid import UUID, uuid4
@dataclass
class User:
id: UUID
email: str
name: str
is_active: bool = True
class UserRepository(ABC):
@abstractmethod
def save(self, user: User) -> User:
pass
@abstractmethod
def get_by_id(self, user_id: UUID) -> Optional[User]:
pass
@abstractmethod
def get_by_email(self, email: str) -> Optional[User]:
pass
@abstractmethod
def list_all(self) -> List[User]:
pass
@abstractmethod
def delete(self, user_id: UUID) -> bool:
pass
Implementazione con SQLAlchemy
from sqlalchemy import create_engine, Column, String, Boolean
from sqlalchemy.dialects.postgresql import UUID as PGUUID
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
Base = declarative_base()
class UserModel(Base):
__tablename__ = 'users'
id = Column(PGUUID(as_uuid=True), primary_key=True)
email = Column(String, unique=True, nullable=False)
name = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
class SQLAlchemyUserRepository(UserRepository):
def __init__(self, session: Session):
self.session = session
def save(self, user: User) -> User:
user_model = UserModel(
id=user.id,
email=user.email,
name=user.name,
is_active=user.is_active
)
self.session.add(user_model)
self.session.commit()
return user
def get_by_id(self, user_id: UUID) -> Optional[User]:
user_model = self.session.query(UserModel).filter(
UserModel.id == user_id
).first()
if not user_model:
return None
return User(
id=user_model.id,
email=user_model.email,
name=user_model.name,
is_active=user_model.is_active
)
def get_by_email(self, email: str) -> Optional[User]:
user_model = self.session.query(UserModel).filter(
UserModel.email == email
).first()
if not user_model:
return None
return User(
id=user_model.id,
email=user_model.email,
name=user_model.name,
is_active=user_model.is_active
)
def list_all(self) -> List[User]:
users = self.session.query(UserModel).all()
return [
User(
id=u.id,
email=u.email,
name=u.name,
is_active=u.is_active
)
for u in users
]
def delete(self, user_id: UUID) -> bool:
result = self.session.query(UserModel).filter(
UserModel.id == user_id
).delete()
self.session.commit()
return result > 0
Repository in Memoria per i Test
class InMemoryUserRepository(UserRepository):
def __init__(self):
self.users: dict[UUID, User] = {}
def save(self, user: User) -> User:
self.users[user.id] = user
return user
def get_by_id(self, user_id: UUID) -> Optional[User]:
return self.users.get(user_id)
def get_by_email(self, email: str) -> Optional[User]:
for user in self.users.values():
if user.email == email:
return user
return None
def list_all(self) -> List[User]:
return list(self.users.values())
def delete(self, user_id: UUID) -> bool:
if user_id in self.users:
del self.users[user_id]
return True
return False
Strato dei Servizi: Orchestrare la Logica Aziendale
Lo Strato dei Servizi implementa i casi d’uso e orchestra il flusso tra i repository, i servizi esterni e la logica del dominio.
from typing import Optional
from uuid import uuid4
class UserAlreadyExistsError(Exception):
pass
class UserNotFoundError(Exception):
pass
class UserService:
def __init__(
self,
user_repository: UserRepository,
email_service: EmailSender,
event_publisher: 'EventPublisher'
):
self.user_repository = user_repository
self.email_service = email_service
self.event_publisher = event_publisher
def register_user(self, email: str, name: str) -> User:
# Verifica se l'utente esiste
existing_user = self.user_repository.get_by_email(email)
if existing_user:
raise UserAlreadyExistsError(f"L'utente con email {email} esiste già")
# Crea nuovo utente
user = User(
id=uuid4(),
email=email,
name=name,
is_active=True
)
# Salva nel repository
user = self.user_repository.save(user)
# Invia email di benvenuto
self.email_service.send(
to=user.email,
subject="Benvenuto!",
body=f"Ciao {user.name}, benvenuto sulla nostra piattaforma!"
)
# Pubblica evento
self.event_publisher.publish('user.registered', {
'user_id': str(user.id),
'email': user.email
})
return user
def deactivate_user(self, user_id: UUID) -> User:
user = self.user_repository.get_by_id(user_id)
if not user:
raise UserNotFoundError(f"Utente {user_id} non trovato")
user.is_active = False
user = self.user_repository.save(user)
self.event_publisher.publish('user.deactivated', {
'user_id': str(user.id)
})
return user
Iniezione delle Dipendenze in Python
La natura dinamica di Python rende l’iniezione delle dipendenze semplice senza richiedere framework pesanti.
Iniezione nel Costruttore
class OrderService:
def __init__(
self,
order_repository: 'OrderRepository',
payment_processor: PaymentProcessor,
notification_service: 'NotificationService'
):
self.order_repository = order_repository
self.payment_processor = payment_processor
self.notification_service = notification_service
def place_order(self, order_data: dict):
# Utilizza le dipendenze iniettate
pass
Contenitore di Dipendenze Semplice
from typing import Dict, Type, Callable, Any
class Container:
def __init__(self):
self._services: Dict[Type, Callable] = {}
self._singletons: Dict[Type, Any] = {}
def register(self, interface: Type, factory: Callable):
self._services[interface] = factory
def register_singleton(self, interface: Type, instance: Any):
self._singletons[interface] = instance
def resolve(self, interface: Type):
if interface in self._singletons:
return self._singletons[interface]
factory = self._services.get(interface)
if factory:
return factory(self)
raise ValueError(f"Nessuna registrazione trovata per {interface}")
# Utilizzo
def create_container() -> Container:
container = Container()
# Registra i servizi
container.register_singleton(
Session,
sessionmaker(bind=create_engine('postgresql://...'))()
)
container.register(
UserRepository,
lambda c: SQLAlchemyUserRepository(c.resolve(Session))
)
container.register(
EmailSender,
lambda c: SMTPEmailSender()
)
container.register(
UserService,
lambda c: UserService(
c.resolve(UserRepository),
c.resolve(EmailSender),
c.resolve(EventPublisher)
)
)
return container
Architettura Esagonale (Porte e Adattatori)
L’Architettura Esagonale colloca la logica aziendale al centro con gli adattatori che gestiscono la comunicazione esterna.
Definizione delle Porte (Interfacce)
# Porta di Input (Primaria)
class CreateUserUseCase(Protocol):
def execute(self, request: 'CreateUserRequest') -> 'CreateUserResponse':
...
# Porta di Output (Secondaria)
class UserPersistencePort(Protocol):
def save(self, user: User) -> User:
...
def find_by_email(self, email: str) -> Optional[User]:
...
Implementazione degli Adattatori
from pydantic import BaseModel, EmailStr
# Adattatore di Input (API REST)
from fastapi import FastAPI, Depends, HTTPException
class CreateUserRequest(BaseModel):
email: EmailStr
name: str
class CreateUserResponse(BaseModel):
id: str
email: str
name: str
app = FastAPI()
@app.post("/users", response_model=CreateUserResponse)
def create_user(
request: CreateUserRequest,
user_service: UserService = Depends(get_user_service)
):
try:
user = user_service.register_user(
email=request.email,
name=request.name
)
return CreateUserResponse(
id=str(user.id),
email=user.email,
name=user.name
)
except UserAlreadyExistsError as e:
raise HTTPException(status_code=400, detail=str(e))
# Adattatore di Output (Database)
# Già implementato come SQLAlchemyUserRepository
Pattern di Progettazione Orientati al Dominio
Oggetti Valore
Oggetti immutabili definiti dai loro attributi:
from dataclasses import dataclass
from typing import Pattern
import re
@dataclass(frozen=True)
class Email:
value: str
EMAIL_PATTERN: Pattern = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')
def __post_init__(self):
if not self.EMAIL_PATTERN.match(self.value):
raise ValueError(f"Email non valida: {self.value}")
def __str__(self):
return self.value
@dataclass(frozen=True)
class Money:
amount: float
currency: str
def __post_init__(self):
if self.amount < 0:
raise ValueError("L'importo non può essere negativo")
if self.currency not in ['USD', 'EUR', 'GBP']:
raise ValueError(f"Valuta non supportata: {self.currency}")
def add(self, other: 'Money') -> 'Money':
if self.currency != other.currency:
raise ValueError("Non è possibile aggiungere valute diverse")
return Money(self.amount + other.amount, self.currency)
Aggregati
Gruppo di oggetti del dominio trattati come unità singola:
from dataclasses import dataclass, field
from typing import List
from datetime import datetime
@dataclass
class OrderItem:
product_id: UUID
quantity: int
price: Money
def total(self) -> Money:
return Money(
self.price.amount * self.quantity,
self.price.currency
)
@dataclass
class Order:
id: UUID
customer_id: UUID
items: List[OrderItem] = field(default_factory=list)
status: str = "pending"
created_at: datetime = field(default_factory=datetime.now)
def add_item(self, product_id: UUID, quantity: int, price: Money):
item = OrderItem(product_id, quantity, price)
self.items.append(item)
def remove_item(self, product_id: UUID):
self.items = [
item for item in self.items
if item.product_id != product_id
]
def total(self) -> Money:
if not self.items:
return Money(0, "USD")
return sum(
(item.total() for item in self.items),
Money(0, self.items[0].price.currency)
)
def confirm(self):
if not self.items:
raise ValueError("Non è possibile confermare un ordine vuoto")
if self.status != "pending":
raise ValueError("L'ordine è già stato processato")
self.status = "confirmed"
Eventi del Dominio
Gli eventi del dominio abilitano il decoupling tra i componenti e supportano architetture orientate agli eventi. Per sistemi a larga scala orientati agli eventi, considera l’implementazione di streaming degli eventi con servizi come AWS Kinesis—vedi Costruire Microservizi Orientati agli Eventi con AWS Kinesis per una guida dettagliata.
from dataclasses import dataclass
from datetime import datetime
from typing import List, Callable
@dataclass
class DomainEvent:
occurred_at: datetime = field(default_factory=datetime.now)
@dataclass
class OrderConfirmed(DomainEvent):
order_id: UUID
customer_id: UUID
total: Money
class EventPublisher:
def __init__(self):
self._handlers: Dict[Type, List[Callable]] = {}
def subscribe(self, event_type: Type, handler: Callable):
if event_type not in self._handlers:
self._handlers[event_type] = []
self._handlers[event_type].append(handler)
def publish(self, event: DomainEvent):
event_type = type(event)
handlers = self._handlers.get(event_type, [])
for handler in handlers:
handler(event)
Funzionalità moderne di Python per un’architettura pulita
Le funzionalità moderne di Python rendono l’implementazione di un’architettura pulita più elegante e sicura nei tipi. Se hai bisogno di un riferimento rapido per la sintassi e le funzionalità di Python, consulta il Python Cheatsheet.
Hint sui tipi e protocolli
from typing import Protocol, runtime_checkable
@runtime_checkable
class Serializable(Protocol):
def to_dict(self) -> dict:
...
@classmethod
def from_dict(cls, data: dict) -> 'Serializable':
...
def serialize(obj: Serializable) -> dict:
return obj.to_dict()
Pydantic per la validazione
from pydantic import BaseModel, Field, validator
from typing import Optional
class CreateUserDTO(BaseModel):
email: EmailStr
name: str = Field(..., min_length=2, max_length=100)
age: Optional[int] = Field(None, ge=0, le=150)
@validator('name')
def name_must_not_contain_numbers(cls, v):
if any(char.isdigit() for char in v):
raise ValueError('Il nome non può contenere numeri')
return v
class Config:
frozen = True # Rendilo immutabile
Async/Await per le operazioni I/O
La sintassi async/await di Python è particolarmente potente per le operazioni I/O-bound nell’architettura pulita, permettendo interazioni non bloccanti con database e servizi esterni. Quando si distribuiscono applicazioni Python su piattaforme serverless, comprendere le caratteristiche di prestazione diventa cruciale—vedi AWS lambda performance: JavaScript vs Python vs Golang per informazioni sull’ottimizzazione delle funzioni serverless Python.
from typing import List
import asyncio
class AsyncUserRepository(ABC):
@abstractmethod
async def save(self, user: User) -> User:
pass
@abstractmethod
async def get_by_id(self, user_id: UUID) -> Optional[User]:
pass
class AsyncUserService:
def __init__(self, repository: AsyncUserRepository):
self.repository = repository
async def register_user(self, email: str, name: str) -> User:
user = User(id=uuid4(), email=email, name=name)
return await self.repository.save(user)
async def get_users_batch(self, user_ids: List[UUID]) -> List[User]:
tasks = [self.repository.get_by_id(uid) for uid in user_ids]
results = await asyncio.gather(*tasks)
return [u for u in results if u is not None]
Best Practices per la Struttura del Progetto
Un’organizzazione corretta del progetto è essenziale per mantenere un’architettura pulita. Prima di impostare la struttura del progetto, assicurati di utilizzare ambienti virtuali Python per l’isolamento delle dipendenze. Il venv Cheatsheet copre tutto ciò che devi sapere per gestire gli ambienti virtuali. Per progetti moderni Python, considera l’utilizzo di uv - Nuovo Gestore di Pacchetti, Progetti e Ambienti Python, che fornisce una gestione più rapida dei pacchetti e un’installazione dei progetti più veloce.
my_application/
├── domain/ # Regole aziendali
│ ├── __init__.py
│ ├── entities/
│ │ ├── __init__.py
│ │ ├── user.py
│ │ └── order.py
│ ├── value_objects/
│ │ ├── __init__.py
│ │ ├── email.py
│ │ └── money.py
│ ├── events/
│ │ ├── __init__.py
│ │ └── user_events.py
│ └── exceptions.py
│
├── application/ # Regole aziendali dell'applicazione
│ ├── __init__.py
│ ├── use_cases/
│ │ ├── __init__.py
│ │ ├── create_user.py
│ │ └── place_order.py
│ ├── services/
│ │ ├── __init__.py
│ │ └── user_service.py
│ └── ports/
│ ├── __init__.py
│ ├── repositories.py
│ └── external_services.py
│
├── infrastructure/ # Interfacce esterne
│ ├── __init__.py
│ ├── persistence/
│ │ ├── __init__.py
│ │ ├── sqlalchemy/
│ │ │ ├── models.py
│ │ │ └── repositories.py
│ │ └── mongodb/
│ │ └── repositories.py
│ ├── messaging/
│ │ ├── __init__.py
│ │ └── rabbitmq_publisher.py
│ ├── external_services/
│ │ ├── __init__.py
│ │ └── email_service.py
│ └── config.py
│
├── presentation/ # Layer UI/API
│ ├── __init__.py
│ ├── api/
│ │ ├── __init__.py
│ │ ├── dependencies.py
│ │ ├── routes/
│ │ │ ├── __init__.py
│ │ │ ├── users.py
│ │ │ └── orders.py
│ │ └── schemas/
│ │ ├── __init__.py
│ │ └── user_schemas.py
│ └── cli/
│ └── commands.py
│
├── tests/
│ ├── unit/
│ ├── integration/
│ └── e2e/
│
├── main.py # Punto di ingresso dell'applicazione
├── container.py # Configurazione dell'iniezione delle dipendenze
├── pyproject.toml
└── README.md
Testing dell’Architettura Pulita
Testing Unitario della Logica del Dominio
import pytest
from uuid import uuid4
def test_user_creation():
user = User(
id=uuid4(),
email="test@example.com",
name="Test User"
)
assert user.email == "test@example.com"
assert user.is_active is True
def test_order_total_calculation():
order = Order(id=uuid4(), customer_id=uuid4())
order.add_item(
uuid4(),
quantity=2,
price=Money(10.0, "USD")
)
order.add_item(
uuid4(),
quantity=1,
price=Money(5.0, "USD")
)
assert order.total().amount == 25.0
Testing di Integrazione con il Repository
@pytest.fixture
def in_memory_repository():
return InMemoryUserRepository()
def test_user_repository_save_and_retrieve(in_memory_repository):
user = User(
id=uuid4(),
email="test@example.com",
name="Test User"
)
saved_user = in_memory_repository.save(user)
retrieved_user = in_memory_repository.get_by_id(user.id)
assert retrieved_user is not None
assert retrieved_user.email == user.email
Testing del Layer dei Servizi
from unittest.mock import Mock
def test_user_registration():
# Arrange
mock_repository = Mock(spec=UserRepository)
mock_repository.get_by_email.return_value = None
mock_repository.save.return_value = User(
id=uuid4(),
email="test@example.com",
name="Test"
)
mock_email = Mock(spec=EmailSender)
mock_events = Mock(spec=EventPublisher)
service = UserService(mock_repository, mock_email, mock_events)
# Act
user = service.register_user("test@example.com", "Test")
# Assert
assert user.email == "test@example.com"
mock_repository.save.assert_called_once()
mock_email.send.assert_called_once()
mock_events.publish.assert_called_once()
Errori Comuni e Come Evitarli
Over-Engineering
Non implementare un’architettura pulita per applicazioni CRUD semplici. Inizia semplice e ristruttura quando la complessità cresce.
Astrazioni Perdenti
Assicurati che le entità del dominio non contengano annotazioni del database o codice specifico del framework:
# Male
from sqlalchemy import Column
@dataclass
class User:
id: Column(Integer, primary_key=True) # Framework che si infila nel dominio
# Bene
@dataclass
class User:
id: UUID # Oggetto puro del dominio
Dipendenze Circolari
Utilizza l’iniezione delle dipendenze e le interfacce per rompere le dipendenze circolari tra i livelli.
Ignorare il Contesto
L’architettura pulita non è adatta a tutti. Adatta la rigidità dei livelli in base alla dimensione del progetto e all’esperienza del team.
Link Utili
- Clean Architecture di Robert C. Martin
- Documentazione Python Type Hints
- Documentazione Pydantic
- Documentazione Ufficiale di FastAPI
- Documentazione ORM di SQLAlchemy
- Libreria Dependency Injector
- Riferimento su Domain-Driven Design
- Architettura Patterns con Python
- Blog di Martin Fowler sull’Architettura
- Guida ai Pattern di Progettazione in Python
- Python Cheatsheet
- venv Cheatsheet
- uv - Nuovo Gestore di Pacchetti, Progetti e Ambienti Python
- Prestazioni di AWS Lambda: JavaScript vs Python vs Golang
- Costruzione di Microservizi Event-Driven con AWS Kinesis