Arquitectura backend¶
Este documento describe cómo se organizan las piezas del backend para que Django y FastAPI convivan sin duplicar lógica.
Decisión central: Django y FastAPI comparten base de datos y lógica¶
Django es el dueño de los modelos y las migraciones. FastAPI consume una capa de servicios compartida. Ninguno reimplementa la lógica del otro.
Django FastAPI
│ │
├──► ORM ├──► Endpoints API
│ │ │
│ ▼ ▼
│ PostgreSQL ◄──────► core/services/
│ ▲ │
│ │ │
└─── migraciones │
│
┌─────────┴─────────┐
▼ ▼
Django Admin Apps móviles
¿Quién hace qué?¶
Django¶
1. Define los modelos en code/core/models/
2. Corre migraciones: python manage.py makemigrations && migrate
3. Expone Django Admin para operación interna
4. Provee vistas auxiliares si FastAPI las necesita
5. Gestiona la carga de archivos a MinIO
FastAPI¶
1. Recibe requests de las apps móviles y web
2. Valida entrada con Pydantic
3. Llama a funciones de core/services/
4. Devuelve respuestas JSON
5. NO toca la base de datos directamente (salvo queries de lectura)
6. NO corre migraciones
Capa de servicios compartidos¶
core/
__init__.py
models/
__init__.py
user.py
locker_code.py
kyc.py
purchase_report.py
physical_package.py
shipment.py
consolidation.py
quote.py
payment.py
charge.py
incident.py
notification.py
referral.py
credit.py
promo_card.py
file_attachment.py
audit_log.py
services/
__init__.py
auth_service.py
locker_service.py # Genera identificador con dígito verificador
kyc_service.py # Submit, approve, reject
package_service.py # Receive, identify, associate
reconciliation_service.py
consolidation_service.py
quote_service.py # Genera cotización
shipment_service.py
payment_service.py # Procesa pago, aplica crédito
charge_service.py # Calcula cargos por almacenaje
incident_service.py
notification_service.py
referral_service.py # Genera crédito por referido
credit_service.py # Balance, expiry, apply
card_service.py # Emite tarjetas
file_service.py # Upload a MinIO, URL firmada
repositories/
__init__.py
user_repository.py
package_repository.py
...
validators/
locker_validator.py
kyc_validator.py
package_validator.py
permissions/
roles.py
package_permissions.py
Ejemplo: recibir un paquete¶
# core/services/package_service.py
def receive_package(locker_code: str, tracking: str, operator_id: int) -> PhysicalPackage:
# Bloquear con Redis para evitar race conditions
with redis.lock(f"package:receive:{locker_code}", timeout=10):
# Validar identificador
locker = locker_service.get_by_code(locker_code)
if not locker:
raise InvalidLockerCodeError()
# Buscar prealerta por tracking
report = purchase_report_service.find_by_tracking(tracking)
# Crear paquete físico
package = PhysicalPackage.objects.create(
locker_code=locker,
tracking=tracking,
purchase_report=report,
status='recibido_identificado',
received_by=operator_id,
)
# Registrar evento en timeline
package_service.record_event(package, 'recibido_identificado', operator_id)
# Notificar al usuario
notification_service.send(
user=locker.user,
type='package_received',
title='Paquete recibido',
body=f'Tu paquete con tracking {tracking} llegó al almacén.',
)
return package
# FastAPI endpoint
@router.post("/packages/receive")
def receive_package_endpoint(data: ReceivePackageSchema, user=Depends(get_current_operator)):
try:
package = package_service.receive_package(
locker_code=data.locker_code,
tracking=data.tracking,
operator_id=user.id,
)
return PackageResponse.from_orm(package)
except InvalidLockerCodeError:
raise HTTPException(400, "INVALID_LOCKER_CODE")
Sincronización con Redis¶
Redis evita condiciones de carrera en procesos críticos.
Locks¶
import redis
from contextlib import contextmanager
redis_client = redis.Redis()
@contextmanager
def distributed_lock(lock_name: str, timeout: int = 10):
lock = redis_client.lock(lock_name, timeout=timeout)
acquired = lock.acquire(blocking=True)
if not acquired:
raise LockNotAcquiredError()
try:
yield
finally:
lock.release()
# Uso en servicios
with distributed_lock(f"locker:generate:{prefix}"):
sequential = get_next_sequential(prefix)
dv = calculate_check_digit(sequential)
code = create_locker_code(prefix, sequential, dv)
Jobs en background¶
# jobs.py — usando Celery o similar
@celery_app.task
def check_storage_expiry():
"""Se ejecuta diariamente. Paquetes vencidos: aplicar cargo."""
packages = PhysicalPackage.objects.filter(
status='en_almacen',
storage_days__gt=FREE_STORAGE_DAYS,
storage_charge_status='pending',
)
for package in packages:
charge_service.apply_storage_charge(package)
notification_service.send(
user=package.locker_code.user,
type='storage_charge_applied',
title='Cargo por almacenaje',
body=f'Se aplicó un cargo de ${package.storage_charge} por {package.storage_days - FREE_STORAGE_DAYS} días extra.',
)
@celery_app.task
def expire_credits():
"""Se ejecuta diariamente. Crédito vencido: marcar como expirado."""
credits = CreditBalance.objects.filter(
status='active',
expires_at__lt=now(),
)
for credit in credits:
credit_service.expire(credit)
@celery_app.task
def recalculate_levels():
"""Se ejecuta cada hora. Recalcula nivel de cada usuario."""
for user in User.objects.filter(is_active=True):
referral_service.recalculate_level(user)
Seguridad de archivos (MinIO)¶
Subida directa con URL prefirmada¶
1. App solicita URL prefirmada a FastAPI
2. FastAPI genera URL PUT con caducidad (ej. 15 minutos)
3. App sube archivo directo a MinIO
4. MinIO guarda en bucket privado
5. App confirma subida a FastAPI
6. FastAPI crea file_attachment en PostgreSQL
Acceso a archivos privados¶
1. App solicita archivo a FastAPI
2. FastAPI verifica permisos (usuario/propietario u operador)
3. FastAPI genera URL GET prefirmada (ej. 5 minutos)
4. FastAPI registra acceso en audit_log
5. App recibe URL temporal y accede directamente a MinIO
Validación de estados¶
Cada servicio verifica el estado antes de operar:
def dispatch_shipment(shipment_id: int, operator_id: int) -> Shipment:
shipment = Shipment.objects.get(id=shipment_id)
if shipment.status != 'pagado':
raise InvalidShipmentStatusError(
f"Shipment status must be 'pagado', got '{shipment.status}'"
)
if shipment.consolidation.kyc_status != 'aprobado':
raise KYCRequiredError()
# Proceder con el despacho
...
Errores centralizados¶
# core/exceptions.py
class AppError(Exception):
code: str
message: str
status_code: int = 400
class InvalidLockerCodeError(AppError):
code = "INVALID_LOCKER_CODE"
message = "El identificador no existe o es inválido"
status_code = 400
class KYCRequiredError(AppError):
code = "KYC_REQUIRED"
message = "KYC aprobado es requerido para esta operación"
status_code = 403
class PackageNotAvailableError(AppError):
code = "PACKAGE_NOT_AVAILABLE"
message = "El paquete no está disponible para esta operación"
status_code = 400
class QuoteExpiredError(AppError):
code = "QUOTE_EXPIRED"
message = "La cotización ha expirado"
status_code = 400
class PaymentFailedError(AppError):
code = "PAYMENT_FAILED"
message = "El pago no pudo procesarse"
status_code = 400
FastAPI convierte estas excepciones en respuestas HTTP:
# fastapi/main.py
from core.exceptions import AppError
@app.exception_handler(AppError)
async def app_error_handler(request, exc: AppError):
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.code, "message": exc.message},
)
Testing¶
tests/
unit/
services/
test_package_service.py
test_kyc_service.py
test_payment_service.py
integration/
test_api_packages.py
test_api_payments.py
fixtures/
users.yaml
packages.yaml
- Unit tests para cada servicio en
core/services/ - Integration tests para endpoints FastAPI
- Django Admin se prueba manualmente o con tests de Selenium