Documentation Index
Fetch the complete documentation index at: https://mintlify.com/farojas85/fast-rest-api/llms.txt
Use this file to discover all available pages before exploring further.
Hexagonal Architecture was chosen for DIAN REST API to keep the billing domain — tax concept validation, voluntary tip exclusion from the taxable base, invoice total reconciliation, and buyer identity — completely isolated from infrastructure details like SOAP envelopes and SQL schemas. Because src/domain/entities/ and src/application/use_cases/ never import from src/infrastructure/, the entire business logic tier can be exercised in pure unit tests with simple Python mocks, no running database and no network call required. When the DIAN changes its WSDL or the database is swapped from PostgreSQL to another engine, only the adapter implementation changes; the domain and use case layers are untouched.
The Three Layers
1. Domain — Pure Entities
src/domain/entities/pedido_local.py contains the canonical model of a restaurant invoice. Every class is a plain Pydantic model with no imports from infrastructure or application. Tax rates, item quantities, payment methods, and buyer data are all expressed here.
from decimal import Decimal
from typing import List
from pydantic import BaseModel, Field
class ConceptoImpuesto(BaseModel):
nombre: str = Field(..., description="Nombre del impuesto, e.g., INC, IVA")
tasa: Decimal = Field(..., description="Tasa del impuesto, e.g., 0.08 para 8%")
monto: Decimal = Field(..., description="Valor calculado del impuesto")
class ItemPedido(BaseModel):
id_producto: str
descripcion: str
cantidad: int
precio_unitario: Decimal
impuestos: List[ConceptoImpuesto] = Field(default_factory=list)
total_item: Decimal
class MetodoPago(BaseModel):
tipo: str = Field(..., description="Efectivo, Tarjeta, Transferencia, etc")
monto: Decimal
class Adquirente(BaseModel):
tipo_documento: str = Field(default="13", description="Tipo de documento (13=CC, 31=NIT, etc.)")
numero_documento: str = Field(default="222222222222", description="Número de identificación")
razon_social: str = Field(default="Consumidor Final", description="Nombre o Razón social")
class PedidoLocal(BaseModel):
id_pedido: str
adquirente: Adquirente = Field(default_factory=Adquirente, description="Datos del cliente")
items: List[ItemPedido]
subtotal: Decimal
propina_voluntaria: Decimal = Field(default=Decimal('0.00'), description="Propina, no hace parte de base gravable")
impuestos_totales: Decimal
total_factura: Decimal
metodos_pago: List[MetodoPago]
# Regla: puramente entidad de dominio, sin base de datos o lógica externa.
PedidoLocal and its nested models have zero imports from src/application/ or src/infrastructure/. This is a hard architectural constraint: the domain layer must never depend on an outer ring.
2. Application — Port Interfaces
The application layer defines the contracts that infrastructure must honor. Two Protocol interfaces act as the outward-facing ports for the use case.
src/application/ports/dian_port.py — the DIAN SOAP output port:
from typing import Protocol, Any, Dict
class IDianSoapPort(Protocol):
"""
Puerto (Interfaz) de salida para la comunicación con la DIAN.
La implementación (Zeep o HTTPX) reside en infraestructura.
"""
async def send_invoice(self, invoice_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Envía una factura electrónica a la DIAN.
"""
...
async def send_credit_note(self, note_data: Dict[str, Any]) -> Dict[str, Any]:
"""
Envía una nota crédito a la DIAN.
"""
...
src/application/ports/log_repository.py — the audit log output port:
from typing import Protocol, Dict, Any
class IInvoiceLogRepositoryPort(Protocol):
"""
Puerto (Interfaz) para el guardado de los logs de transacciones de facturación.
La implementación (SQLAlchemy) reside en infraestructura.
"""
async def save_log(self, status: str, payload_in: Dict[str, Any], response_out: Dict[str, Any] = None) -> int:
"""
Guarda un nuevo registro o actualiza el estado de la transacción.
"""
...
typing.Protocol definitions, meaning any class that implements the required async methods satisfies the interface — no inheritance required. This enables lightweight mock adapters in tests without subclassing anything.
3. Infrastructure — Adapters
Driving adapters (primary) — FastAPI controllers translate inbound HTTP requests into use case calls. The controller in src/infrastructure/controllers/pedido_local_controller.py deserializes the request body into PedidoLocalCreateRequest, calls use_case.execute(payload), and maps the result to an HTTP response.
Driven adapters (secondary) — DianSoapAdapter in src/infrastructure/adapters/dian_soap/dian_adapter.py implements IDianSoapPort. It constructs a zeep.AsyncClient backed by an httpx.AsyncClient with a 30-second timeout, then calls service.SendTestSetAsync(fileName, contentFile, testSetId) against the DIAN WSDL. A future PostgreSQL adapter will implement IInvoiceLogRepositoryPort using SQLAlchemy async sessions.
Use Case: FacturarPedidoLocalUseCase
The use case in src/application/use_cases/facturar_pedido_local.py receives both ports through constructor injection and orchestrates the full billing flow:
class FacturarPedidoLocalUseCase:
"""
Caso de uso para registrar y facturar un pedido de consumo local.
Sigue directrices de Arquitectura Hexagonal.
"""
def __init__(
self,
dian_port: IDianSoapPort,
log_repository: IInvoiceLogRepositoryPort
):
self.dian_port = dian_port
self.log_repository = log_repository
async def execute(self, payload: PedidoLocalCreateRequest) -> PedidoLocalResponse:
"""
1. Validar reglas de negocio (ej. totales vs items, propinas no en base gravable).
2. Mapear DTO a Entidad de Dominio.
3. Registrar intento de factura en DB (Estado: Procesando).
4. Llamar adaptador DIAN SOAP.
5. Actualizar Log según respuesta.
6. Retornar DTO.
"""
# Step 1 — Business validation
expected_total = payload.subtotal + payload.impuestos_totales + payload.propina_voluntaria
if payload.total_factura != expected_total:
raise ValueError("El total de la factura no coincide con la suma del subtotal, impuestos y propina.")
# Step 2 — Map DTO to domain entities
items_dominio = [
ItemPedido(
id_producto=item.id_producto,
descripcion=item.descripcion,
cantidad=item.cantidad,
precio_unitario=item.precio_unitario,
impuestos=[
ConceptoImpuesto(
nombre=imp.nombre,
tasa=imp.tasa,
monto=imp.monto
) for imp in item.impuestos
],
total_item=item.total_item
) for item in payload.items
]
metodos_pago = [MetodoPago(tipo=m.tipo, monto=m.monto) for m in payload.metodos_pago]
adquirente_dominio = Adquirente(
tipo_documento=payload.adquirente.tipo_documento,
numero_documento=payload.adquirente.numero_documento,
razon_social=payload.adquirente.razon_social
) if payload.adquirente else Adquirente()
pedido_domain = PedidoLocal(
id_pedido=payload.id_pedido_origen,
adquirente=adquirente_dominio,
items=items_dominio,
subtotal=payload.subtotal,
propina_voluntaria=payload.propina_voluntaria,
impuestos_totales=payload.impuestos_totales,
total_factura=payload.total_factura,
metodos_pago=metodos_pago
)
# Step 3 — Log attempt (PROCESANDO)
await self.log_repository.save_log(status="PROCESANDO", payload_in=payload.model_dump())
# Steps 4–6 — Call DIAN, update log, return DTO
try:
dian_res = await self.dian_port.send_invoice(invoice_data=pedido_domain.model_dump())
track_id = dian_res.get("track_id", "UNKNOWN")
cufe = dian_res.get("cufe")
status = dian_res.get("status", "SUCCESS")
await self.log_repository.save_log(
status=status,
payload_in=payload.model_dump(),
response_out=dian_res
)
return PedidoLocalResponse(
track_id=track_id,
cufe=cufe,
status=status,
message="Factura procesada con éxito."
)
except Exception as e:
await self.log_repository.save_log(
status="FAILED",
payload_in=payload.model_dump(),
response_out={"error": str(e)}
)
raise e
execute docstring map directly to the sequence diagram in the Data Flow page: validate → map → log PROCESANDO → call DIAN port → log SUCCESS/FAILED → return response DTO.
Dependency Injection
FastAPI’s Depends() system wires the concrete DianSoapAdapter and the use case together at request time. Both provider functions live in the controller module:
def get_dian_adapter() -> DianSoapAdapter:
"""Provee la instancia configurada del Adaptador SOAP de la DIAN."""
# En producción se decidiría entre WSDL de Habilitación o Producción
# según un flag de entorno (ej. settings.ENVIRONMENT == 'production')
wsdl = settings.DIAN_WSDL_URL_HABILITACION if settings.ENVIRONMENT != 'production' else settings.DIAN_WSDL_URL_PRODUCCION
return DianSoapAdapter(
wsdl_url=wsdl,
cert_path=str(settings.DIAN_CERT_PATH),
password=settings.DIAN_CERT_PASSWORD.get_secret_value()
)
def get_use_case(
dian_adapter: DianSoapAdapter = Depends(get_dian_adapter)
) -> FacturarPedidoLocalUseCase:
"""
Inyecta las dependencias al caso de uso.
Queda pendiente Mock de InvoiceLogRepositoryPort hasta el siguiente Issue de BD.
"""
class MockLogRepo:
async def log_attempt(self, **kwargs): pass
async def update_status(self, **kwargs): pass
return FacturarPedidoLocalUseCase(
dian_port=dian_adapter,
log_repo=MockLogRepo()
)
MockLogRepo inside get_use_case() is a temporary no-op placeholder for the log port while the real PostgreSQL adapter (SQLAlchemy async) is pending implementation. The constructor call passes log_repo=MockLogRepo(), but FacturarPedidoLocalUseCase.__init__ declares the parameter as log_repository — this keyword argument name mismatch will raise a TypeError at runtime and must be corrected to log_repository=MockLogRepo() before the dependency injection chain is exercised. Once the database adapter is complete, MockLogRepo will be replaced with the real IInvoiceLogRepositoryPort implementation.
Depends(get_use_case):
@router.post("/", response_model=PedidoLocalResponse, status_code=status.HTTP_201_CREATED)
async def facturar_pedido_local(
payload: PedidoLocalCreateRequest,
use_case: FacturarPedidoLocalUseCase = Depends(get_use_case)
):
result = await use_case.execute(payload)
...
Testing Benefits
Because FacturarPedidoLocalUseCase only holds references to the two port interfaces (not concrete adapters), tests in tests/application/use_cases/test_facturar_pedido_local.py replace both with in-memory mocks:
class MockDianPort:
async def send_invoice(self, invoice_data: dict) -> dict:
return {"track_id": "ABC-123", "status": "SUCCESS", "message": "OK"}
async def send_credit_note(self, note_data: dict) -> dict:
pass
class MockLogRepositoryPort:
def __init__(self):
self.logs = []
self.counter = 1
async def save_log(self, status: str, payload_in: dict, response_out: dict = None) -> int:
self.logs.append({"id": self.counter, "status": status, "payload_in": payload_in, "response_out": response_out})
self.counter += 1
return self.counter - 1
@pytest.mark.asyncio
async def test_facturar_pedido_local_success(mock_dian_port, mock_log_repository, valid_payload):
use_case = FacturarPedidoLocalUseCase(mock_dian_port, mock_log_repository)
response = await use_case.execute(valid_payload)
assert response.track_id == "ABC-123"
assert response.status == "SUCCESS"
assert len(mock_log_repository.logs) == 2 # 1 for PROCESANDO, 1 for SUCCESS
@pytest.mark.asyncio
async def test_facturar_pedido_local_invalid_totals(mock_dian_port, mock_log_repository, valid_payload):
invalid_payload = valid_payload.model_copy()
invalid_payload.total_factura = Decimal("9999.00") # Deliberately wrong
use_case = FacturarPedidoLocalUseCase(mock_dian_port, mock_log_repository)
with pytest.raises(ValueError, match="El total de la factura no coincide"):
await use_case.execute(invalid_payload)
MockDianPort returns a hardcoded success dict and MockLogRepositoryPort records every save_log call in memory. The success test verifies both the returned track_id and that exactly two log entries were written (one PROCESANDO, one SUCCESS). The invalid-totals test verifies the ValueError is raised before any log or SOAP call. Zero network connections, zero database sessions.