Driving Adapters: API con FastAPI
El adapter HTTP traduce la petición al lenguaje de la aplicación.
Un driving adapter (o adapter primario) es el componente que activa la aplicación desde el exterior. En el contexto de una API web, el driving adapter es el conjunto formado por el router, los esquemas Pydantic y las funciones controladoras que reciben una petición HTTP y la convierten en una llamada al caso de uso correspondiente. Su responsabilidad es traducir el protocolo HTTP al lenguaje de la aplicación, no ejecutar lógica de negocio.
En FastAPI, el driving adapter se articula en tres piezas. El router registra las rutas y delega en funciones controladoras. Los schemas Pydantic definen los DTOs de entrada y salida: validan tipos, formatos y restricciones de la petición HTTP. La función controladora extrae los datos del request, construye un Command o Query, lo despacha al handler correspondiente y mapea el resultado a un response schema. Ese es el flujo completo de un driving adapter bien formado.
La separación entre el schema Pydantic (DTO HTTP) y los objetos del dominio es fundamental. El schema Pydantic puede tener campos opcionales, alias para el JSON, validadores de formato de cadena, o campos que el frontend necesita pero el dominio no. El Value Object del dominio refleja el lenguaje del negocio y protege invariantes. Si haces que el schema herede del dominio o viceversa, acoples el protocolo HTTP a la lógica de negocio: un cambio en el contrato HTTP obliga a modificar el dominio.
El controlador no debe contener lógica de negocio por ningún motivo. Si ves un `if`/`else` basado en datos del negocio dentro de un controlador, esa lógica pertenece al Aggregate o al Application Service. El controlador puede tener lógica de presentación trivial —mapear un código de error de dominio a un código HTTP, serializar la respuesta— pero nunca decidir qué hacer con los datos según reglas del negocio. Esta disciplina mantiene el controlador testeable con un mock del handler y sin necesidad de base de datos.
Los schemas Pydantic viven en la capa de infraestructura junto con el router y los controladores, no en el dominio ni en la aplicación. El dominio tiene Value Objects; la aplicación tiene Commands y Queries; la infraestructura HTTP tiene schemas Pydantic. Cada capa tiene su propio contrato de datos y el mapping entre ellos ocurre en el controlador. Esta separación permite versionar el API sin tocar el dominio y cambiar el dominio sin romper el contrato HTTP.
# infrastructure/api/schemas.py — HTTP DTOs (Pydantic)
# Never inherit from domain entities or value objects.
from pydantic import BaseModel, Field
from uuid import UUID
class <DoSomething>Request(BaseModel):
<field_one>: str = Field(..., min_length=1, max_length=200)
<field_two>: UUID
class <DoSomething>Response(BaseModel):
id: UUID
<result_field>: str
# infrastructure/api/controllers.py — Driving Adapter
# Translates HTTP → Command → Response. No business logic.
from fastapi import HTTPException, status
from ...application.commands.<do_something>.command import <DoSomething>Command
from ...application.commands.<do_something>.handler import <DoSomething>Handler
from .schemas import <DoSomething>Request, <DoSomething>Response
async def <do_something>_controller(
body: <DoSomething>Request,
handler: <DoSomething>Handler,
) -> <DoSomething>Response:
command = <DoSomething>Command(
<field_one>=body.<field_one>,
<field_two>=body.<field_two>,
)
# BAD: do NOT add if/else domain logic here
# if body.<field_one>.startswith('forbidden'): # <-- domain rule in controller
# raise HTTPException(status_code=400, detail='...')
try:
result = handler.handle(command)
except <DomainError> as exc:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
return <DoSomething>Response(id=result.id, <result_field>=result.<result_field>)
# infrastructure/api/router.py — Route registration only
from fastapi import APIRouter, Depends
from .controllers import <do_something>_controller
router = APIRouter(prefix='/<resource>', tags=['<resource>'])
router.post('/', response_model=<DoSomething>Response)(<do_something>_controller)Debugging lab
Detecta y corrige el error o la violación de diseño.
- 6.2.5.1
# infrastructure/api/controllers.py async def create_<resource>_controller( body: Create<Resource>Request, handler: Create<Resource>Handler, ) -> Create<Resource>Response: if body.<field> == '<special_value>': body.<field> = '<default_value>' # business default if not body.<other_field>.startswith('<prefix>'): raise HTTPException(status_code=400, detail='Invalid prefix') command = Create<Resource>Command(**body.model_dump()) result = handler.handle(command) return Create<Resource>Response(id=result.id)
- 6.2.5.2
# infrastructure/api/controllers.py from sqlalchemy.orm import Session async def update_<resource>_controller( resource_id: UUID, body: Update<Resource>Request, session: Session, ) -> Update<Resource>Response: row = session.get(<AggregateORM>, str(resource_id)) row.<field> = body.<field> session.commit() return Update<Resource>Response(id=resource_id)
- 6.2.5.3
# domain/model/<aggregate>.py from pydantic import BaseModel class <Aggregate>(BaseModel): # domain entity inheriting Pydantic id: UUID <field>: str # infrastructure/api/schemas.py from ...domain.model.<aggregate> import <Aggregate> class <Aggregate>Response(<Aggregate>): # schema inheriting domain entity pass
- 6.2.5.4
# application/commands/<do_something>/handler.py from fastapi import HTTPException class <DoSomething>Handler: def handle(self, command: <DoSomething>Command) -> None: aggregate = self._repo.get(command.aggregate_id) if aggregate is None: raise HTTPException(status_code=404, detail='Not found')
- 6.2.5.5
# infrastructure/api/controllers.py async def get_<resource>_controller( resource_id: UUID, session: Session, # session injected directly into controller ) -> <Resource>Response: row = session.query(<AggregateORM>).filter_by(id=str(resource_id)).first() if not row: raise HTTPException(status_code=404) return <Resource>Response(id=row.id, <field>=row.<field>)