DEV Community: Enrique Lazo Bello

🚀 Notificaciones en Tiempo Real con SSE, FastAPI y Celery

Enrique Lazo Bello — Mon, 16 Mar 2026 15:44:48 +0000

¿Alguna vez has hecho clic en un botón, la página se queda "pensando" y no sabes si algo está pasando? Para evitar esto en procesos largos —generar un reporte, procesar imágenes, ejecutar un análisis— necesitas comunicación en tiempo real. Pero más importante: necesitas elegir la herramienta correcta.

¿Cuándo usar SSE y cuándo no?

Antes de escribir una línea de código, la decisión de arquitectura más importante es elegir el protocolo correcto.

Criterio	Polling	WebSockets	SSE
Dirección	Client → Server	Bidireccional	Server → Client
Complejidad	Baja	Alta	Baja/Media
Reconexión automática	Manual	Manual	Nativa
Soporte HTTP/2	Sí	No	Sí
Ideal para	Datos muy esporádicos	Chat, juegos	Progreso, notificaciones

Regla práctica: Si el cliente nunca necesita enviar datos por el mismo canal abierto, SSE gana. Es HTTP estándar, tiene reconexión automática incorporada en el navegador, y es significativamente más simple de escalar que WebSockets.

La Arquitectura: Por qué 4 piezas y no 2

La tentación inicial es hacer todo en un solo proceso FastAPI. El problema: FastAPI (ASGI) está optimizado para I/O, no para cómputo pesado. Si pones una tarea de 30 segundos directamente en un endpoint, bloqueas el event loop y degradas toda la API.

✅ Arquitectura correcta:
Browser → FastAPI → Redis (cola) → Celery Worker → Redis (estado)
                 ↑                                      |
                 └──────── SSE stream ←─────────────────┘

Cada pieza tiene una responsabilidad única:

FastAPI: Recibe peticiones HTTP y mantiene las conexiones SSE abiertas.
Celery Worker: Ejecuta las tareas pesadas en un proceso separado.
Redis: Actúa como Broker (cola) y como State Store (donde vive el progreso).
React: Consume el stream SSE y actualiza la UI.

Paso 1: El Backend (FastAPI)

El endpoint de encolar

Usamos celery_app.send_task para enviar la tarea al worker sin esperar a que termine.

@app.post("/task")
async def create_task():
    task_id = str(uuid.uuid4())
    # Encolar la tarea y guardar estado inicial
    celery_app.send_task("tasks.run_simulation", args=[task_id], task_id=task_id)
    redis_client.set(f"task:{task_id}", json.dumps({"status": "pending", "progress": 0}))
    return {"task_id": task_id}

El stream SSE: Máxima estabilidad con `StreamingResponse`

Para evitar errores de encoding y asegurar compatibilidad total, usamos StreamingResponse formateando manualmente el protocolo SSE (data: <json>\n\n).

from fastapi.responses import StreamingResponse

@app.get("/task/{task_id}/stream")
async def task_stream(task_id: str, request: Request):
    async def event_generator():
        while True:
            # 1. Detectar si el usuario cerró la pestaña
            if await request.is_disconnected():
                break

            # 2. Leer progreso de Redis
            raw = redis_client.get(f"task:{task_id}")
            if raw:
                data = json.loads(raw)
                data["id"] = task_id
                # 3. Formato obligatorio SSE: "data: ...\n\n"
                yield f"data: {json.dumps(data)}\n\n"

                if data.get("status") == "done":
                    break

            await asyncio.sleep(0.5)

    return StreamingResponse(
        event_generator(), 
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}
    )

Paso 2: El Worker (Celery)

El worker es el encargado de actualizar Redis en cada fase de la simulación.

@celery_app.task(name="tasks.run_simulation", bind=True)
def run_simulation(self, task_id):
    # Fase 1: Simulación de inicio
    time.sleep(2)
    redis_client.set(f"task:{task_id}", json.dumps({"status": "processing", "progress": 33}))

    # Fase 2: Trabajo pesado
    time.sleep(random.uniform(3, 6))
    redis_client.set(f"task:{task_id}", json.dumps({"status": "processing", "progress": 66}))

    # Fase 3: Finalización
    result = {
        "status": "done", 
        "progress": 100, 
        "result": {"metric": random.uniform(0, 100), "processed_at": "..."}
    }
    redis_client.set(f"task:{task_id}", json.dumps(result))
    return result

Paso 3: El Frontend (React)

Usamos la API nativa EventSource. Un detalle clave: cerramos la conexión manualmente cuando el estado es done para liberar recursos del servidor.

function TaskCard({ task, onUpdate }: TaskCardProps) {
  useEffect(() => {
    if (task.status === "done") return;

    const eventSource = new EventSource(`http://localhost:8000/task/${task.id}/stream`);

    eventSource.onmessage = (event) => {
      const updated = JSON.parse(event.data);
      onUpdate(updated);
      if (updated.status === "done") eventSource.close();
    };

    return () => eventSource.close(); // Cleanup
  }, [task.id, task.status]);

  // ... Render barrita de progreso y badges
}

Paso 4: Dockerización

Orquestamos los 4 servicios para que la red interna permita la comunicación entre ellos usando sus nombres de servicio (redis, api, etc.).

services:
  redis:
    image: redis:7-alpine
  api:
    build: ./api
    environment:
      - REDIS_URL=redis://redis:6379/0
    ports:
      - "8000:8000"
  worker:
    build: ./worker
    environment:
      - REDIS_URL=redis://redis:6379/0
  frontend:
    build: ./frontend
    ports:
      - "5173:5173"

Casos de Producción: Lo que no te dicen

Nginx Buffering: Si usas Nginx como proxy, debes desactivar el buffering (proxy_buffering off), de lo contrario, los eventos SSE se acumularán y llegarán todos de golpe al final.
Límite de conexiones: Los navegadores limitan a 6 las conexiones SSE por dominio en HTTP/1.1. Si tu app escala, considera usar HTTP/2.
Manejo de desconexiones: El check request.is_disconnected() en FastAPI es vital para no dejar bucles infinitos en el backend.

Conclusión

Este stack (FastAPI + Celery + Redis + SSE) es una solución de grado industrial. Es robusta, fácil de debuguear y escala mucho mejor que intentar manejar sockets bidireccionales cuando solo necesitas enviar progreso al usuario.

¿Has implementado sistemas en tiempo real antes? ¿Qué retos encontraste con los streams de datos? ¡Te leo en los comentarios! 👇

¿Cómo diseño APIs REST limpias y fáciles de mantener (Python/Laravel)?

Enrique Lazo Bello — Sat, 06 Dec 2025 12:37:57 +0000

Durante los últimos años trabajando como arquitecto backend, he visto equipos completos bloqueados por APIs mal diseñadas. El problema nunca es la complejidad técnica del negocio, sino decisiones estructurales que tomamos al inicio y que terminan convirtiéndose en deuda técnica imposible de pagar. Una API bien diseñada no es aquella que funciona hoy, es aquella que otro desarrollador puede extender mañana sin necesidad de llamarte a las 3 AM.

En proyectos con multiples API o microservicios, he aplicado un conjunto de principios que transforman APIs caóticas en sistemas predecibles. No se trata de arquitecturas complejas ni patrones rebuscados. Se trata de decisiones pequeñas y consistentes que facilitan el mantenimiento, reducen bugs en producción y permiten que el equipo escale sin inconvenientes.

Este artículo desglosa mi enfoque práctico para estructurar APIs REST profesionales, con ejemplos reales de decisiones que marcan la diferencia entre un sistema mantenible y uno que nadie quiere tocar.

Estructurando nuestro proyecto

La primera regla de una API mantenible es que cualquier desarrollador pueda predecir dónde está cada cosa sin necesidad de buscar. He trabajado en sistemas donde cada endpoint tenía su propia lógica de organización: algunos mezclaban validación con lógica de negocio, otros tenían consultas SQL directamente en los controladores. El resultado era predecible: cada cambio requería horas de análisis y el riesgo de romper algo no relacionado era altísimo.

La arquitectura que aplico divide el flujo en tres capas con responsabilidades claras:

Controlador (Capa de Transporte): Recibe el HTTP Request, orquesta la llamada al servicio y devuelve el HTTP Response. No contiene reglas de negocio. Su única responsabilidad es traducir entre el protocolo HTTP y tu dominio de aplicación.

Servicio (Lógica de Negocio): Aquí reside la inteligencia del sistema. Es agnóstico al framework HTTP, no sabe si la petición vino de una API REST, una tarea programada o un comando de consola. Esta independencia permite reutilizar la lógica en múltiples contextos sin duplicación.

Repositorio (Capa de Persistencia): Abstrae las consultas a la base de datos o servicios externos. Si mañana cambias de MySQL a PostgreSQL o consumes una API externa, el servicio no debería enterarse.

Mi estructura estándar en FastAPI:

app/
├── api/
│   └── v1/
│       ├── endpoints/
│       │   ├── users.py
│       │   └── orders.py
│       └── __init__.py
├── schemas/
│   ├── user.py
│   └── order.py
├── services/
│   ├── user_service.py
│   └── order_service.py
└── repositories/
    ├── user_repository.py
    └── order_repository.py

En Laravel, mantengo el mismo principio con la estructura nativa del framework:

app/
├── Http/
│   ├── Controllers/
│   │   └── Api/
│   │       └── V1/
│   │           ├── UserController.php
│   │           └── OrderController.php
│   └── Requests/
│       └── V1/
│           ├── CreateUserRequest.php
│           └── CreateOrderRequest.php
├── Services/
│   ├── UserService.php
│   └── OrderService.php
└── Repositories/
    ├── UserRepository.php
    └── OrderRepository.php

Esta separación tiene un propósito claro: cada capa tiene una única razón para cambiar. Cuando necesitas modificar cómo se valida un campo, sabes exactamente que vas a schemas o Requests. Cuando cambias la lógica de cálculo de descuentos, vas directo al servicio correspondiente. Cuando optimizas una consulta SQL, modificas solo el repositorio sin tocar la lógica de negocio.

En un proyecto de e-commerce que heredé, refactoricé una API donde todo estaba en los controladores. El resultado: reducimos el tiempo promedio de implementación de nuevas features de 3 días a 1 día, y los bugs en producción cayeron un 60% en el primer trimestre. La razón fue simple: al separar responsabilidades, cada cambio afectaba solo a una capa, eliminando efectos secundarios inesperados.

Validar todo desde el lado del cliente

Uno de los errores más costosos que veo en APIs es validar datos dentro de la lógica de negocio. Cuando la validación está dispersa, terminas con código defensivo por todas partes, verificando el mismo campo en múltiples lugares y generando inconsistencias. La validación debe ocurrir en la puerta de entrada. Si el dato no es válido, recházalo inmediatamente con HTTP 422 antes de consumir recursos del servidor.

En FastAPI, Pydantic es tu mejor aliado:

from pydantic import BaseModel, EmailStr, validator, Field
from typing import Optional
from datetime import datetime

class CreateUserRequest(BaseModel):
    email: EmailStr
    full_name: str = Field(..., min_length=3, max_length=100)
    phone: Optional[str] = None
    birth_date: datetime

    @validator('full_name')
    def validate_full_name(cls, v):
        if len(v.strip()) < 3:
            raise ValueError('El nombre debe tener al menos 3 caracteres')
        return v.strip()

    @validator('phone')
    def validate_phone(cls, v):
        if v and not v.isdigit():
            raise ValueError('El teléfono solo puede contener dígitos')
        return v

    @validator('birth_date')
    def validate_age(cls, v):
        age = (datetime.now() - v).days / 365
        if age < 18:
            raise ValueError('El usuario debe ser mayor de edad')
        return v

@router.post("/users", status_code=201)
async def create_user(
    user_data: CreateUserRequest,
    service: UserService = Depends(get_user_service)
):
    user = await service.create_user(user_data)
    return {"user_id": user.id, "email": user.email}

El servicio recibe datos ya validados y transformados. No necesita verificar si el email es válido o si el nombre cumple las reglas de negocio. Esto permite que las pruebas unitarias del servicio se concentren en la lógica real, no en casos extremos de validación.

En Laravel, los FormRequests cumplen el mismo rol y extraen la validación a una clase dedicada que limpia el controlador:

class CreateUserRequest extends FormRequest
{
    public function authorize(): bool
    {
        return true;
    }

    public function rules(): array
    {
        return [
            'email' => 'required|email|unique:users,email|max:255',
            'full_name' => 'required|min:3|max:100',
            'phone' => 'nullable|digits:9',
            'birth_date' => 'required|date|before:18 years ago',
        ];
    }

    public function messages(): array
    {
        return [
            'birth_date.before' => 'El usuario debe ser mayor de edad',
            'phone.digits' => 'El teléfono debe tener 9 dígitos',
            'email.unique' => 'Este email ya está registrado',
        ];
    }

    protected function prepareForValidation()
    {
        $this->merge([
            'full_name' => trim($this->full_name),
        ]);
    }
}

Esta centralización tiene un beneficio enorme cuando tu API crece: si cambias las reglas de validación de usuarios, modificas un único archivo. No tienes que rastrear validaciones dispersas por 15 controladores diferentes. Además, la documentación automática en FastAPI (Swagger) y las respuestas de validación en Laravel son consistentes y predecibles para los consumidores de tu API.

Versionado simple

El versionado de APIs genera debates interminables sobre la estrategia perfecta. Después de implementar sistemas con versionado en headers, query params y URLs, mi recomendación es usar versionado en la URL con prefijo /v1, /v2. Es explícito, fácil de documentar, y no requiere que los clientes de tu API configuren headers especiales. Cuando un cliente consume api.tuapp.com/v1/users, sabe exactamente qué versión está usando.

Mi regla práctica para crear una nueva versión:

Cambios backward-compatible (agregar campos opcionales, nuevos endpoints): mantener misma versión.

Cambios breaking (eliminar campos, cambiar tipos de datos, modificar comportamiento esperado): nueva versión.

El error común es versionar desde el día 1 sin justificación. Prefiero versionar cuando toca, no antes. Si tu API aún no está en producción o solo la consumes internamente, no necesitas versiones múltiples. La complejidad que agrega el versionado prematuro no vale la pena.

En FastAPI:

from fastapi import APIRouter

router_v1 = APIRouter(prefix="/api/v1", tags=["v1"])
router_v2 = APIRouter(prefix="/api/v2", tags=["v2"])

# v1: retorna user_id como string y estructura simple
@router_v1.get("/users/{user_id}")
async def get_user_v1(user_id: str):
    return {
        "user_id": user_id,
        "name": "John Doe",
        "email": "john@example.com"
    }

# v2: retorna user_id como int, agrega metadata y estructura más rica
@router_v2.get("/users/{user_id}")
async def get_user_v2(user_id: int):
    return {
        "user_id": user_id,
        "name": "John Doe",
        "email": "john@example.com",
        "metadata": {
            "created_at": "2024-01-01T00:00:00Z",
            "last_login": "2024-12-01T10:30:00Z"
        },
        "preferences": {
            "language": "es",
            "timezone": "America/Lima"
        }
    }

# Registro en la app
app.include_router(router_v1)
app.include_router(router_v2)

En Laravel:

// routes/api.php
Route::prefix('v1')->group(function () {
    Route::get('/users/{id}', [UserController::class, 'show']);
    Route::post('/users', [UserController::class, 'store']);
});

Route::prefix('v2')->group(function () {
    Route::get('/users/{id}', [UserControllerV2::class, 'show']);
    Route::post('/users', [UserControllerV2::class, 'store']);
});

Una estrategia que aplico en producción: cuando lanzamos v2, v1 permanece activa con un header X-API-Deprecation-Date que informa la fecha de cierre del API. Esto permite a los clientes migrar sin prisa, y a nosotros deprecar versiones antiguas de manera controlada. Típicamente mantengo una versión antigua activa entre 6 y 12 meses después de lanzar la nueva, dependiendo de la criticidad de la API y el número de integraciones activas.

Manejo explícito de errores: hablar el idioma correcto

Un error HTTP 500 con el mensaje "Error" no ayuda a nadie. He depurado sistemas donde cada problema se devolvía como 200 OK con un campo status: "error" en el JSON. Esto rompe el contrato HTTP y obliga a los clientes a implementar lógica adicional para interpretar respuestas. El protocolo HTTP ya tiene códigos de estado bien definidos, úsalos correctamente.

Mi estructura estándar de errores en FastAPI:

from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from datetime import datetime

class APIException(Exception):
    def __init__(self, status_code: int, message: str, code: str = None, details: dict = None):
        self.status_code = status_code
        self.message = message
        self.code = code or "INTERNAL_ERROR"
        self.details = details or {}

@app.exception_handler(APIException)
async def api_exception_handler(request: Request, exc: APIException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "code": exc.code,
                "message": exc.message,
                "details": exc.details,
                "path": str(request.url),
                "timestamp": datetime.now().isoformat(),
                "request_id": request.state.request_id
            }
        }
    )

# Uso en el servicio
class UserService:
    async def get_user(self, user_id: int):
        user = await self.repository.find(user_id)
        if not user:
            raise APIException(
                status_code=404,
                code="USER_NOT_FOUND",
                message="Usuario no encontrado",
                details={"user_id": user_id}
            )
        return user

    async def create_user(self, user_data: CreateUserRequest):
        try:
            return await self.repository.create(user_data)
        except IntegrityError:
            raise APIException(
                status_code=409,
                code="EMAIL_ALREADY_EXISTS",
                message="El email ya está registrado",
                details={"email": user_data.email}
            )

En Laravel, los Custom Exceptions permiten un patrón similar:

class UserNotFoundException extends Exception
{
    public function render($request)
    {
        return response()->json([
            'error' => [
                'code' => 'USER_NOT_FOUND',
                'message' => 'Usuario no encontrado',
                'details' => ['user_id' => $request->route('id')],
                'path' => $request->path(),
                'timestamp' => now()->toIso8601String(),
                'request_id' => request()->header('X-Request-ID')
            ]
        ], 404);
    }
}

class EmailAlreadyExistsException extends Exception
{
    protected $email;

    public function __construct($email)
    {
        parent::__construct();
        $this->email = $email;
    }

    public function render($request)
    {
        return response()->json([
            'error' => [
                'code' => 'EMAIL_ALREADY_EXISTS',
                'message' => 'El email ya está registrado',
                'details' => ['email' => $this->email],
                'path' => $request->path(),
                'timestamp' => now()->toIso8601String(),
                'request_id' => request()->header('X-Request-ID')
            ]
        ], 409);
    }
}

Esta consistencia permite que los equipos frontend construyan manejadores de errores genéricos. Si todas las respuestas de error siguen la misma estructura, el cliente puede implementar un interceptor único que procese cualquier error de la API. Además, el campo code permite al frontend reaccionar programáticamente a errores específicos sin tener que parsear mensajes de texto que pueden cambiar.

Logging y trazabilidad: debugging sin adivinar

Cuando un endpoint falla en producción a las 2 AM, necesitas responder una pregunta: qué pasó exactamente. Sin logging estructurado, estás adivinando. No hablo de logs masivos, hablo de logs útiles que permitan reconstruir el flujo completo de una request en minutos.

Mi configuración base en FastAPI con middleware de logging:

import structlog
from uuid import uuid4
from fastapi import Request
import time

logger = structlog.get_logger()

@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    request_id = str(uuid4())
    request.state.request_id = request_id

    structlog.contextvars.clear_contextvars()
    structlog.contextvars.bind_contextvars(
        request_id=request_id,
        path=request.url.path,
        method=request.method,
        client_ip=request.client.host
    )

    start_time = time.time()

    logger.info("request_started")

    try:
        response = await call_next(request)
        duration = time.time() - start_time

        logger.info(
            "request_completed",
            status_code=response.status_code,
            duration_ms=round(duration * 1000, 2)
        )

        response.headers["X-Request-ID"] = request_id
        return response
    except Exception as e:
        duration = time.time() - start_time
        logger.error(
            "request_failed",
            error=str(e),
            duration_ms=round(duration * 1000, 2),
            exc_info=True
        )
        raise

# En tu servicio
class UserService:
    async def create_user(self, user_data: CreateUserRequest):
        logger.info("creating_user", email=user_data.email)

        try:
            user = await self.repository.create(user_data)
            logger.info("user_created", user_id=user.id)
            return user
        except IntegrityError as e:
            logger.error("user_creation_failed", error=str(e), email=user_data.email)
            raise APIException(409, "EMAIL_ALREADY_EXISTS", "El email ya existe")

En Laravel:

// app/Http/Middleware/LogRequests.php
class LogRequests
{
    public function handle(Request $request, Closure $next)
    {
        $requestId = Str::uuid()->toString();
        $request->headers->set('X-Request-ID', $requestId);

        $startTime = microtime(true);

        Log::channel('api')->info('request_started', [
            'request_id' => $requestId,
            'method' => $request->method(),
            'url' => $request->fullUrl(),
            'ip' => $request->ip(),
            'user_id' => optional($request->user())->id,
        ]);

        try {
            $response = $next($request);

            $duration = microtime(true) - $startTime;

            Log::channel('api')->info('request_completed', [
                'request_id' => $requestId,
                'duration_ms' => round($duration * 1000, 2),
                'status_code' => $response->getStatusCode(),
            ]);

            return $response->header('X-Request-ID', $requestId);
        } catch (Exception $e) {
            $duration = microtime(true) - $startTime;

            Log::channel('api')->error('request_failed', [
                'request_id' => $requestId,
                'duration_ms' => round($duration * 1000, 2),
                'error' => $e->getMessage(),
                'trace' => $e->getTraceAsString(),
            ]);

            throw $e;
        }
    }
}

// app/Services/UserService.php
class UserService
{
    public function createUser(CreateUserRequest $request)
    {
        $requestId = request()->header('X-Request-ID');

        Log::withContext(['request_id' => $requestId]);
        Log::info('creating_user', ['email' => $request->email]);

        try {
            $user = $this->repository->create($request->validated());
            Log::info('user_created', ['user_id' => $user->id]);
            return $user;
        } catch (QueryException $e) {
            Log::error('user_creation_failed', [
                'error' => $e->getMessage(),
                'email' => $request->email
            ]);
            throw new EmailAlreadyExistsException($request->email);
        }
    }
}

Este request_id es oro puro en debugging. Cuando un cliente reporta un error, me da el X-Request-ID, y yo puedo rastrear cada log relacionado a esa petición específica. En un sistema con 10,000 requests por minuto, esto reduce el tiempo de diagnóstico de horas a minutos. Además, incluir duración en milisegundos permite detectar endpoints lentos y optimizarlos proactivamente.

Evitar sobrecargar los endpoints: una responsabilidad por endpoint

Uno de los antipatrones más comunes es el endpoint Frankenstein: hace validación, lógica de negocio, envío de emails, actualización de caché y notificaciones push. Todo en un solo lugar. Si un endpoint hace demasiado, mezcla casos de uso o devuelve sets de datos gigantes, lo divido sin excepción.

Ejemplo de lo que no debes hacer:

@router.post("/orders")
async def create_order_bad(order_data: dict):
    # Validación manual mezclada
    if not order_data.get('items'):
        raise HTTPException(400, "Items requeridos")

    # Lógica de negocio directa
    total = sum(item['price'] * item['qty'] for item in order_data['items'])
    if total > 10000:
        order_data['requires_approval'] = True

    # Guardado directo sin abstracción
    order = db.orders.insert(order_data)

    # Efectos secundarios sincrónicos
    send_email(order_data['email'], "Orden creada")
    cache.set(f"order:{order.id}", order)
    push_service.notify(order_data['user_id'], "Tu orden está en proceso")

    return order

Esto es una bomba de tiempo. Cada responsabilidad agregada aumenta la complejidad de testing y la probabilidad de efectos secundarios. Si el envío de email falla, toda la creación de la orden falla. Si necesitas agregar una nueva acción cuando se crea una orden, tienes que modificar este endpoint y arriesgarte a romper algo.

La versión correcta:

@router.post("/orders", status_code=201)
async def create_order(
    order_data: CreateOrderRequest,
    service: OrderService = Depends(get_order_service)
):
    """
    Crea una nueva orden de compra.

    La orden se crea de forma síncrona.
    Las notificaciones y procesos secundarios se ejecutan de forma asíncrona.
    """
    order = await service.create_order(order_data)
    return {"order_id": order.id, "status": order.status}

# En el servicio
class OrderService:
    async def create_order(self, order_data: CreateOrderRequest):
        # La validación ya ocurrió en CreateOrderRequest
        logger.info("creating_order", user_id=order_data.user_id)

        order = await self.repository.create(order_data)

        # Las operaciones asíncronas se delegan a workers
        await self.event_bus.publish(
            OrderCreatedEvent(order_id=order.id, user_id=order_data.user_id)
        )

        logger.info("order_created", order_id=order.id)
        return order

El endpoint solo coordina. La lógica de envío de emails, actualización de caché y notificaciones vive en event handlers separados que procesan el evento OrderCreatedEvent. Si el envío de email falla, no afecta la creación de la orden. Si necesitas agregar una nueva acción cuando se crea una orden, creas un nuevo handler sin tocar el endpoint.

Ejemplo práctico: estructura completa de un endpoint real

Veamos cómo todos estos principios se integran en un endpoint de producción. Este ejemplo muestra la creación de una orden de compra con validación completa, cálculo de totales y manejo de stock.

Estructura completa en FastAPI:

# schemas/order.py
from pydantic import BaseModel, validator, Field
from typing import List
from decimal import Decimal

class OrderItem(BaseModel):
    product_id: int = Field(..., gt=0)
    quantity: int = Field(..., gt=0)

    @validator('quantity')
    def validate_quantity(cls, v):
        if v < 1:
            raise ValueError('La cantidad debe ser mayor a 0')
        if v > 100:
            raise ValueError('La cantidad máxima por producto es 100')
        return v

class CreateOrderRequest(BaseModel):
    user_id: int = Field(..., gt=0)
    items: List[OrderItem] = Field(..., min_items=1)
    shipping_address: str = Field(..., min_length=10, max_length=500)

    @validator('items')
    def validate_items(cls, v):
        if not v:
            raise ValueError('La orden debe tener al menos un producto')
        if len(v) > 50:
            raise ValueError('Máximo 50 productos por orden')
        return v

# repositories/order_repository.py
class OrderRepository:
    def __init__(self, db):
        self.db = db

    async def create(self, order_data: dict):
        query = """
            INSERT INTO orders (user_id, total, status, shipping_address, created_at)
            VALUES (:user_id, :total, :status, :shipping_address, NOW())
            RETURNING id, user_id, total, status, shipping_address, created_at
        """
        result = await self.db.execute(query, order_data)
        return result

    async def create_items(self, order_id: int, items: List[dict]):
        query = """
            INSERT INTO order_items (order_id, product_id, quantity, price)
            VALUES (:order_id, :product_id, :quantity, :price)
        """
        await self.db.execute_many(query, items)

# services/order_service.py
class OrderService:
    def __init__(
        self,
        repository: OrderRepository,
        product_service: ProductService,
        event_bus: EventBus
    ):
        self.repository = repository
        self.product_service = product_service
        self.event_bus = event_bus

    async def create_order(self, order_data: CreateOrderRequest):
        logger.info("creating_order", user_id=order_data.user_id, items_count=len(order_data.items))

        # Validar disponibilidad y calcular total
        total = Decimal(0)
        validated_items = []

        for item in order_data.items:
            product = await self.product_service.get_product(item.product_id)

            if product.stock < item.quantity:
                raise APIException(
                    400,
                    "INSUFFICIENT_STOCK",
                    "Stock insuficiente",
                    {
                        "product_id": item.product_id,
                        "requested": item.quantity,
                        "available": product.stock
                    }
                )

            item_total = product.price * item.quantity
            total += item_total

            validated_items.append({
                "product_id": item.product_id,
                "quantity": item.quantity,
                "price": product.price
            })

        # Crear orden
        order = await self.repository.create({
            "user_id": order_data.user_id,
            "total": total,
            "status": "pending",
            "shipping_address": order_data.shipping_address
        })

        # Crear items de la orden
        items_with_order_id = [
            {**item, "order_id": order.id}
            for item in validated_items
        ]
        await self.repository.create_items(order.id, items_with_order_id)

        # Publicar evento para procesos asíncronos
        await self.event_bus.publish(
            OrderCreatedEvent(
                order_id=order.id,
                user_id=order_data.user_id,
                total=float(total)
            )
        )

        logger.info("order_created", order_id=order.id, total=float(total))
        return order

# endpoints/orders.py
@router.post("/orders", status_code=201, response_model=OrderResponse)
async def create_order(
    order_data: CreateOrderRequest,
    service: OrderService = Depends(get_order_service),
    current_user: User = Depends(get_current_user)
):
    """
    Crea una nueva orden de compra.

    Valida stock disponible, calcula el total y genera la orden.
    Las notificaciones y emails se procesan de forma asíncrona.

    Requiere autenticación.
    """
    if order_data.user_id != current_user.id:
        raise APIException(
            403,
            "FORBIDDEN",
            "No tienes permiso para crear órdenes de otros usuarios"
        )

    order = await service.create_order(order_data)

    return {
        "order_id": order.id,
        "total": float(order.total),
        "status": order.status,
        "created_at": order.created_at.isoformat()
    }

Implementación equivalente en Laravel:

// app/Http/Requests/CreateOrderRequest.php
class CreateOrderRequest extends FormRequest
{
    public function authorize(): bool
    {
        return $this->user()->id === $this->input('user_id');
    }

    public function rules(): array
    {
        return [
            'user_id' => 'required|integer|exists:users,id',
            'items' => 'required|array|min:1|max:50',
            'items.*.product_id' => 'required|integer|exists:products,id',
            'items.*.quantity' => 'required|integer|min:1|max:100',
            'shipping_address' => 'required|string|min:10|max:500',
        ];
    }
}

// app/Repositories/OrderRepository.php
class OrderRepository
{
    public function create(array $data): Order
    {
        return DB::transaction(function () use ($data) {
            $order = Order::create([
                'user_id' => $data['user_id'],
                'total' => $data['total'],
                'status' => $data['status'],
                'shipping_address' => $data['shipping_address'],
            ]);

            foreach ($data['items'] as $item) {
                $order->items()->create($item);
            }

            return $order->fresh('items');
        });
    }
}

// app/Services/OrderService.php
class OrderService
{
    public function __construct(
        private OrderRepository $repository,
        private ProductService $productService
    ) {}

    public function createOrder(array $data): Order
    {
        $requestId = request()->header('X-Request-ID');
        Log::withContext(['request_id' => $requestId]);

        Log::info('creating_order', [
            'user_id' => $data['user_id'],
            'items_count' => count($data['items'])
        ]);

        $total = 0;
        $validatedItems = [];

        foreach ($data['items'] as $item) {
            $product = $this->productService->getProduct($item['product_id']);

            if ($product->stock < $item['quantity']) {
                throw new InsufficientStockException(
                    $item['product_id'],
                    $item['quantity'],
                    $product->stock
                );
            }

            $itemTotal = $product->price * $item['quantity'];
            $total += $itemTotal;

            $validatedItems[] = [
                'product_id' => $item['product_id'],
                'quantity' => $item['quantity'],
                'price' => $product->price,
            ];
        }

        $order = $this->repository->create([
            'user_id' => $data['user_id'],
            'total' => $total,
            'status' => 'pending',
            'shipping_address' => $data['shipping_address'],
            'items' => $validatedItems,
        ]);

        event(new OrderCreated($order));

        Log::info('order_created', [
            'order_id' => $order->id,
            'total' => $total
        ]);

        return $order;
    }
}

// app/Http/Controllers/Api/V1/OrderController.php
class OrderController extends Controller
{
    public function __construct(private OrderService $service) {}

    public function store(CreateOrderRequest $request): JsonResponse
    {
        $order = $this->service->createOrder($request->validated());

        return response()->json([
            'order_id' => $order->id,
            'total' => $order->total,
            'status' => $order->status,
            'created_at' => $order->created_at->toIso8601String(),
        ], 201);
    }
}

Este endpoint cumple con todos los principios discutidos: estructura predecible con cada componente en su lugar, validación temprana antes de llegar al servicio, manejo explícito de errores con códigos y mensajes claros, logging estructurado con request_id para trazabilidad completa, y responsabilidad única donde el endpoint solo coordina mientras el servicio contiene toda la lógica.

Conclusión

Diseñar APIs mantenibles no requiere arquitecturas complejas ni frameworks exóticos. Requiere disciplina y consistencia en decisiones pequeñas que tomamos cada día. La estructura predecible reduce la carga cognitiva del equipo. La validación temprana elimina bugs antes de que lleguen a producción. El versionado explícito protege a tus clientes de cambios inesperados. El manejo de errores claro acelera el debugging. El logging estructurado convierte el diagnóstico en ciencia, no en arte adivinatorio.

Cada una de estas prácticas tiene un costo inicial: requiere planificación, coordinación con el equipo y establecer estándares claros. Pero ese costo se paga una sola vez. El beneficio se multiplica con cada nueva feature, cada bug que se evita y cada desarrollador que se incorpora al equipo sin necesidad de semanas de onboarding. La mantenibilidad de una API no depende del framework, sino de la disciplina arquitectónica.

¿Cuál ha sido tu mayor desafío al mantener APIs en producción?
¿Qué prácticas te han funcionado mejor en tu equipo?
Déjamelo en los comentarios.

Contabilidad para Django Developers: Implementando Reportería Regulatoria SBS

Enrique Lazo Bello — Sat, 09 Nov 2024 01:08:14 +0000

Introducción

Como desarrollador Django, ¿alguna vez te has enfrentado al desafío de implementar un sistema contable regulatorio? Si la palabra "contabilidad" te genera la misma ansiedad que un deployment fallido en producción, este tutorial es para ti.

Aprenderás a implementar un sistema de reportería regulatoria básica para la SBS (Superintendencia de Banca y Seguros) utilizando Django, traduciendo conceptos contables a términos de programación que ya conoces. Al final, podrás:

Crear modelos Django para manejar estados financieros
Implementar validaciones automáticas de cuadre contable
Gestionar operaciones en múltiples monedas
Todo esto usando únicamente el admin de Django

Prerrequisitos

Django>=5.0
python-decimal>=3.12

Conocimientos necesarios:

Experiencia básica con Django y su ORM
Familiaridad con modelos Django y el admin
Conocimientos básicos de Python 3

Conceptos Clave para Developers

La Contabilidad como Sistema de Base de Datos

Piensa en la contabilidad como un sistema de base de datos con estas características:

ACID compliant (como PostgreSQL)
Double-entry (cada transacción afecta dos o más registros)
Event sourcing (cada movimiento queda registrado)

# Analogía entre conceptos
class ContabilidadConceptos:
    """
    Conceptos contables traducidos a términos de programación:

    Cuenta Contable    -> Tabla
    Asiento           -> Transacción
    Debe/Haber        -> Crédito/Débito (como git add/remove)
    Balance           -> Checkpoint/Snapshot
    """
    pass

Implementación Práctica

1. Modelos Base

from decimal import Decimal
from django.db import models
from django.core.exceptions import ValidationError
from django.utils.translation import gettext_lazy as _

class Currency(models.Model):
    code = models.CharField(max_length=3, unique=True)
    name = models.CharField(max_length=50)
    exchange_rate = models.DecimalField(
        max_digits=10, 
        decimal_places=4,
        help_text="Tipo de cambio respecto a moneda local"
    )

    class Meta:
        verbose_name_plural = "Currencies"

    def __str__(self):
        return self.code

class Account(models.Model):
    """Cuenta contable según plan contable SBS"""
    ACCOUNT_TYPES = [
        ('A', 'Activo'),
        ('P', 'Pasivo'),
        ('R', 'Resultado'),
    ]

    code = models.CharField(max_length=6, unique=True)
    name = models.CharField(max_length=100)
    type = models.CharField(max_length=1, choices=ACCOUNT_TYPES)
    currency = models.ForeignKey(Currency, on_delete=models.PROTECT)
    is_analytic = models.BooleanField(
        default=False,
        help_text="Indica si la cuenta acepta movimientos"
    )

    def __str__(self):
        return f"{self.code} - {self.name}"

    def get_balance(self, date=None):
        """Obtiene el saldo de la cuenta a una fecha"""
        query = self.movements.all()
        if date:
            query = query.filter(accounting_date__lte=date)

        debit = query.aggregate(
            total=models.Sum('debit_amount', default=Decimal('0.00'))
        )['total']
        credit = query.aggregate(
            total=models.Sum('credit_amount', default=Decimal('0.00'))
        )['total']

        if self.type in ['A', 'R']:
            return debit - credit
        return credit - debit

    class Meta:
        ordering = ['code']

class AccountingEntry(models.Model):
    """Asiento contable"""
    date = models.DateField()
    number = models.CharField(max_length=10, unique=True)
    description = models.TextField()
    is_adjusted = models.BooleanField(
        default=False,
        help_text="Indica si es un asiento de ajuste"
    )
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    def clean(self):
        """Validación de cuadre contable"""
        if self.pk:  # Solo validamos entradas existentes
            debit_sum = self.movements.aggregate(
                total=models.Sum('debit_amount', default=Decimal('0.00'))
            )['total']
            credit_sum = self.movements.aggregate(
                total=models.Sum('credit_amount', default=Decimal('0.00'))
            )['total']

            if debit_sum != credit_sum:
                raise ValidationError(
                    _('El asiento no cuadra. Débito: %(debit)s, Crédito: %(credit)s'),
                    params={'debit': debit_sum, 'credit': credit_sum},
                )

    class Meta:
        verbose_name_plural = "Accounting Entries"

class AccountingMove(models.Model):
    """Movimiento contable individual"""
    entry = models.ForeignKey(
        AccountingEntry,
        on_delete=models.PROTECT,
        related_name='movements'
    )
    account = models.ForeignKey(
        Account,
        on_delete=models.PROTECT,
        related_name='movements'
    )
    debit_amount = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=Decimal('0.00')
    )
    credit_amount = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=Decimal('0.00')
    )
    exchange_rate = models.DecimalField(
        max_digits=10,
        decimal_places=4,
        help_text="Tipo de cambio usado en la operación"
    )
    accounting_date = models.DateField()

    def clean(self):
        """Validaciones de movimiento"""
        if self.debit_amount and self.credit_amount:
            raise ValidationError(
                _('Un movimiento no puede tener débito y crédito simultáneamente')
            )

        if not self.debit_amount and not self.credit_amount:
            raise ValidationError(
                _('El movimiento debe tener al menos un monto')
            )

        if not self.account.is_analytic:
            raise ValidationError(
                _('La cuenta seleccionada no acepta movimientos')
            )

    def save(self, *args, **kwargs):
        if not self.exchange_rate:
            self.exchange_rate = self.account.currency.exchange_rate
        super().save(*args, **kwargs)

    class Meta:
        ordering = ['accounting_date', 'id']

2. Admin personalizado

from django.contrib import admin
from django.db.models import Sum, Value
from django.db.models.functions import Coalesce

@admin.register(Currency)
class CurrencyAdmin(admin.ModelAdmin):
    list_display = ['code', 'name', 'exchange_rate']
    search_fields = ['code', 'name']

@admin.register(Account)
class AccountAdmin(admin.ModelAdmin):
    list_display = ['code', 'name', 'type', 'currency', 'current_balance']
    list_filter = ['type', 'currency', 'is_analytic']
    search_fields = ['code', 'name']

    def current_balance(self, obj):
        return obj.get_balance()
    current_balance.short_description = 'Saldo Actual'

class AccountingMoveInline(admin.TabularInline):
    model = AccountingMove
    extra = 2
    fields = ['account', 'debit_amount', 'credit_amount', 'exchange_rate', 'accounting_date']

@admin.register(AccountingEntry)
class AccountingEntryAdmin(admin.ModelAdmin):
    list_display = ['number', 'date', 'description', 'is_adjusted', 'total_amount']
    list_filter = ['date', 'is_adjusted']
    search_fields = ['number', 'description']
    inlines = [AccountingMoveInline]

    def total_amount(self, obj):
        return obj.movements.aggregate(
            total=Coalesce(Sum('debit_amount'), Value(0))
        )['total']
    total_amount.short_description = 'Monto Total'

3. Tests Unitarios

import pytest
from decimal import Decimal
from django.core.exceptions import ValidationError
from django.utils import timezone
from .models import Currency, Account, AccountingEntry, AccountingMove

@pytest.mark.django_db
class TestAccountingSystem:
    def setup_method(self):
        # Configuración inicial
        self.pen = Currency.objects.create(
            code='PEN',
            name='Soles',
            exchange_rate=Decimal('1.0000')
        )
        self.usd = Currency.objects.create(
            code='USD',
            name='Dólares',
            exchange_rate=Decimal('3.7500')
        )

        # Cuentas de prueba
        self.cash_pen = Account.objects.create(
            code='101101',
            name='Caja MN',
            type='A',
            currency=self.pen,
            is_analytic=True
        )
        self.cash_usd = Account.objects.create(
            code='101102',
            name='Caja ME',
            type='A',
            currency=self.usd,
            is_analytic=True
        )
        self.income = Account.objects.create(
            code='701101',
            name='Ingresos Financieros',
            type='R',
            currency=self.pen,
            is_analytic=True
        )

    def test_account_balance(self):
        # Crear asiento contable
        entry = AccountingEntry.objects.create(
            date=timezone.now().date(),
            number='AS0001',
            description='Test Entry'
        )

        # Movimientos
        AccountingMove.objects.create(
            entry=entry,
            account=self.cash_pen,
            debit_amount=Decimal('1000.00'),
            accounting_date=entry.date
        )
        AccountingMove.objects.create(
            entry=entry,
            account=self.income,
            credit_amount=Decimal('1000.00'),
            accounting_date=entry.date
        )

        assert self.cash_pen.get_balance() == Decimal('1000.00')
        assert self.income.get_balance() == Decimal('-1000.00')

    def test_unbalanced_entry(self):
        entry = AccountingEntry.objects.create(
            date=timezone.now().date(),
            number='AS0002',
            description='Unbalanced Entry'
        )

        AccountingMove.objects.create(
            entry=entry,
            account=self.cash_pen,
            debit_amount=Decimal('1000.00'),
            accounting_date=entry.date
        )

        with pytest.raises(ValidationError):
            entry.clean()

    def test_foreign_currency_movement(self):
        entry = AccountingEntry.objects.create(
            date=timezone.now().date(),
            number='AS0003',
            description='USD Transaction'
        )

        # Movimiento en USD
        AccountingMove.objects.create(
            entry=entry,
            account=self.cash_usd,
            debit_amount=Decimal('100.00'),
            exchange_rate=Decimal('3.7500'),
            accounting_date=entry.date
        )

        # Contrapartida en PEN
        AccountingMove.objects.create(
            entry=entry,
            account=self.income,
            credit_amount=Decimal('375.00'),
            exchange_rate=Decimal('1.0000'),
            accounting_date=entry.date
        )

        assert self.cash_usd.get_balance() == Decimal('100.00')
        assert self.income.get_balance() == Decimal('-375.00')

Ejemplo Real: Sistema de Transferencias

def transfer_between_accounts(
    from_account: Account,
    to_account: Account,
    amount: Decimal,
    date: date,
    description: str
) -> AccountingEntry:
    """
    Realiza una transferencia entre cuentas, manejando diferentes monedas
    """
    # Validaciones previas
    if not from_account.is_analytic or not to_account.is_analytic:
        raise ValidationError('Las cuentas deben ser analíticas')

    if amount <= 0:
        raise ValidationError('El monto debe ser positivo')

    # Crear asiento
    entry = AccountingEntry.objects.create(
        date=date,
        number=f'TR{timezone.now().strftime("%Y%m%d%H%M%S")}',
        description=description
    )

    # Si las monedas son diferentes, calculamos la conversión
    if from_account.currency != to_account.currency:
        to_amount = amount * (
            to_account.currency.exchange_rate / from_account.currency.exchange_rate
        )
    else:
        to_amount = amount

    # Crear movimientos
    AccountingMove.objects.create(
        entry=entry,
        account=from_account,
        credit_amount=amount,
        exchange_rate=from_account.currency.exchange_rate,
        accounting_date=date
    )

    AccountingMove.objects.create(
        entry=entry,
        account=to_account,
        debit_amount=to_amount,
        exchange_rate=to_account.currency.exchange_rate,
        accounting_date=date
    )

    return entry

Mejores Prácticas

Validaciones de Seguridad
- Usar decimal.Decimal para todos los cálculos monetarios
- Implementar permisos granulares en el admin
- Validar cuadre contable antes de guardar
Manejo de Errores
- Usar transacciones para asegurar atomicidad
- Validar tipos de cambio antes de operaciones
- Implementar logging detallado
Patrones de Diseño
- Repository pattern para consultas complejas
- Factory pattern para creación de asientos
- Observer pattern para auditoría

Conclusión

Este tutorial te ha proporcionado una base sólida para implementar reportería regulatoria en Django. Los conceptos clave aprendidos son:

Modelado de datos contables en Django
Validaciones automáticas de cuadre
Manejo de múltiples monedas
Testing de operaciones contables

Contabilidad para Django Developers: Implementando Libros Contables

Enrique Lazo Bello — Sat, 09 Nov 2024 00:54:35 +0000

Introducción

Como desarrollador de Django, ¿alguna vez te has enfrentado al desafío de implementar un sistema contable y te has sentido abrumado por la terminología financiera? No estás solo. La contabilidad puede parecer un mundo completamente diferente a la programación, pero en realidad, comparten muchos conceptos fundamentales.

En este tutorial, aprenderemos a implementar los libros contables principales (Diario, Mayor y Balance de Comprobación) utilizando Django Admin, relacionando conceptos contables con paradigmas de programación que ya conoces. Por ejemplo, un asiento contable es similar a una transacción en base de datos: ambos deben mantener la integridad y ser atómicos.

Al final de este tutorial, serás capaz de implementar un sistema contable robusto utilizando únicamente Django Admin, con validaciones automáticas y pruebas unitarias que garanticen la integridad de los datos.

Prerrequisitos

Python 3.12+
Django 5.0+

Conceptos Clave

La Contabilidad desde una Perspectiva de Desarrollador

Libro Diario → Piense en ello como su historial de confirmaciones de Git
- Cada asiento es como un commit que registra un cambio en el estado financiero
- Debe ser inmutable (como un commit)
- Tiene timestamp y metadata (autor, descripción)
Libro Mayor → Similar a un agregador de logs por categoría
- Agrupa transacciones por "cuenta" (como agrupar logs por servicio)
- Mantiene saldos actualizados (como un contador en Redis)
Balance de Comprobación → Equivalente a revision de cambios
- Verifica que el sistema esté en un estado consistente
- Debe cumplir reglas matemáticas específicas (como assertions en tests)

Implementación Práctica

Modelos Django

from django.db import models
from django.core.exceptions import ValidationError
from django.db.models import Sum
from decimal import Decimal
import uuid

class Account(models.Model):
    """
    Representa una cuenta contable.
    Similar a una tabla en base de datos donde se agregan o restan valores.
    """
    ACCOUNT_TYPES = [
        ('ASSET', 'Activo'),
        ('LIABILITY', 'Pasivo'),
        ('EQUITY', 'Patrimonio'),
        ('INCOME', 'Ingreso'),
        ('EXPENSE', 'Gasto'),
    ]

    code = models.CharField(max_length=20, unique=True)
    name = models.CharField(max_length=100)
    type = models.CharField(max_length=10, choices=ACCOUNT_TYPES)
    is_active = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return f"{self.code} - {self.name}"

    def get_balance(self, end_date=None):
        """
        Calcula el saldo de la cuenta hasta una fecha específica.
        Similar a un query con filtro temporal.
        """
        entries = JournalEntryLine.objects.filter(account=self)
        if end_date:
            entries = entries.filter(entry__date__lte=end_date)

        debit_sum = entries.aggregate(Sum('debit'))['debit__sum'] or Decimal('0')
        credit_sum = entries.aggregate(Sum('credit'))['credit__sum'] or Decimal('0')

        if self.type in ['ASSET', 'EXPENSE']:
            return debit_sum - credit_sum
        return credit_sum - debit_sum

    class Meta:
        ordering = ['code']

class JournalEntry(models.Model):
    """
    Representa un asiento contable.
    Similar a una transacción en base de datos.
    """
    reference = models.UUIDField(default=uuid.uuid4, editable=False)
    date = models.DateField()
    description = models.TextField()
    is_posted = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)
    created_by = models.ForeignKey('auth.User', on_delete=models.PROTECT)

    def clean(self):
        """
        Validaciones a nivel de asiento.
        Similar a validaciones de integridad en DB.
        """
        if self.is_posted:
            raise ValidationError("No se puede modificar un asiento contabilizado")

        # Verificar balance entre débito y crédito
        total_debit = sum(line.debit for line in self.lines.all())
        total_credit = sum(line.credit for line in self.lines.all())

        if total_debit != total_credit:
            raise ValidationError("El asiento no está balanceado")

    def post(self):
        """
        Contabiliza el asiento.
        Similar a un commit en una transacción.
        """
        self.clean()
        self.is_posted = True
        self.save()

    class Meta:
        verbose_name_plural = "Journal Entries"
        ordering = ['-date', '-created_at']

class JournalEntryLine(models.Model):
    """
    Representa una línea de asiento contable.
    Similar a un detalle de transacción.
    """
    entry = models.ForeignKey(
        JournalEntry, 
        related_name='lines', 
        on_delete=models.CASCADE
    )
    account = models.ForeignKey(
        Account, 
        on_delete=models.PROTECT
    )
    description = models.CharField(max_length=200)
    debit = models.DecimalField(
        max_digits=15, 
        decimal_places=2, 
        default=0
    )
    credit = models.DecimalField(
        max_digits=15, 
        decimal_places=2, 
        default=0
    )

    def clean(self):
        """
        Validaciones a nivel de línea.
        Similar a check constraints en DB.
        """
        if self.debit < 0 or self.credit < 0:
            raise ValidationError("Los montos no pueden ser negativos")

        if self.debit > 0 and self.credit > 0:
            raise ValidationError("Una línea no puede tener débito y crédito")

        if self.debit == 0 and self.credit == 0:
            raise ValidationError("El monto debe ser mayor que cero")

    class Meta:
        ordering = ['id']

class TrialBalance(models.Model):
    """
    Balance de Comprobación.
    Similar a un reporte de health check del sistema.
    """
    date = models.DateField(unique=True)
    is_closed = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)
    created_by = models.ForeignKey('auth.User', on_delete=models.PROTECT)

    def generate(self):
        """
        Genera el balance de comprobación.
        Similar a generar un reporte de estado del sistema.
        """
        if self.is_closed:
            raise ValidationError("Este balance ya está cerrado")

        # Eliminar líneas existentes
        self.lines.all().delete()

        # Generar nuevas líneas
        for account in Account.objects.filter(is_active=True):
            balance = account.get_balance(self.date)
            if balance != 0:
                TrialBalanceLine.objects.create(
                    trial_balance=self,
                    account=account,
                    debit=max(balance, 0),
                    credit=max(-balance, 0)
                )

    def validate(self):
        """
        Valida el balance de comprobación.
        Similar a ejecutar test suite.
        """
        total_debit = self.lines.aggregate(Sum('debit'))['debit__sum'] or 0
        total_credit = self.lines.aggregate(Sum('credit'))['credit__sum'] or 0

        if total_debit != total_credit:
            raise ValidationError(
                f"El balance no cuadra: Débito={total_debit}, Crédito={total_credit}"
            )

    class Meta:
        ordering = ['-date']

class TrialBalanceLine(models.Model):
    """
    Línea de Balance de Comprobación.
    Similar a un resultado individual de test.
    """
    trial_balance = models.ForeignKey(
        TrialBalance,
        related_name='lines',
        on_delete=models.CASCADE
    )
    account = models.ForeignKey(Account, on_delete=models.PROTECT)
    debit = models.DecimalField(max_digits=15, decimal_places=2, default=0)
    credit = models.DecimalField(max_digits=15, decimal_places=2, default=0)

    class Meta:
        ordering = ['account__code']

Configuración del Admin

from django.contrib import admin
from django.urls import reverse
from django.utils.html import format_html
from .models import Account, JournalEntry, JournalEntryLine, TrialBalance, TrialBalanceLine

@admin.register(Account)
class AccountAdmin(admin.ModelAdmin):
    list_display = ['code', 'name', 'type', 'get_balance_display', 'is_active']
    list_filter = ['type', 'is_active']
    search_fields = ['code', 'name']

    def get_balance_display(self, obj):
        balance = obj.get_balance()
        return f"{balance:,.2f}"
    get_balance_display.short_description = 'Balance'

class JournalEntryLineInline(admin.TabularInline):
    model = JournalEntryLine
    extra = 2

    def get_readonly_fields(self, request, obj=None):
        if obj and obj.is_posted:
            return ['account', 'description', 'debit', 'credit']
        return []

@admin.register(JournalEntry)
class JournalEntryAdmin(admin.ModelAdmin):
    list_display = ['reference', 'date', 'description', 'is_posted', 'created_by']
    list_filter = ['is_posted', 'date']
    search_fields = ['reference', 'description']
    readonly_fields = ['reference', 'created_by']
    inlines = [JournalEntryLineInline]

    def save_model(self, request, obj, form, change):
        if not obj.pk:
            obj.created_by = request.user
        super().save_model(request, obj, form, change)

    def get_readonly_fields(self, request, obj=None):
        if obj and obj.is_posted:
            return ['reference', 'date', 'description', 'is_posted', 'created_by']
        return ['reference', 'created_by']

class TrialBalanceLineInline(admin.TabularInline):
    model = TrialBalanceLine
    extra = 0
    readonly_fields = ['account', 'debit', 'credit']
    can_delete = False

    def has_add_permission(self, request, obj=None):
        return False

@admin.register(TrialBalance)
class TrialBalanceAdmin(admin.ModelAdmin):
    list_display = ['date', 'is_closed', 'created_by', 'created_at']
    readonly_fields = ['created_by', 'created_at']
    inlines = [TrialBalanceLineInline]

    def save_model(self, request, obj, form, change):
        if not obj.pk:
            obj.created_by = request.user
        super().save_model(request, obj, form, change)

    def response_add(self, request, obj):
        obj.generate()
        obj.validate()
        return super().response_add(request, obj)

Tests Unitarios

from django.test import TestCase
from django.contrib.auth.models import User
from django.core.exceptions import ValidationError
from decimal import Decimal
from .models import Account, JournalEntry, JournalEntryLine, TrialBalance
from datetime import date

class AccountingTestCase(TestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username='testuser',
            password='testpass'
        )

        # Crear cuentas de prueba
        self.cash = Account.objects.create(
            code='1000',
            name='Cash',
            type='ASSET'
        )
        self.bank = Account.objects.create(
            code='1001',
            name='Bank',
            type='ASSET'
        )
        self.revenue = Account.objects.create(
            code='4000',
            name='Revenue',
            type='INCOME'
        )

    def test_journal_entry_balance(self):
        """Verifica que los asientos contables estén balanceados"""
        entry = JournalEntry.objects.create(
            date=date.today(),
            description='Test Entry',
            created_by=self.user
        )

        # Crear líneas desbalanceadas
        JournalEntryLine.objects.create(
            entry=entry,
            account=self.cash,
            description='Debit line',
            debit=Decimal('100.00')
        )

        JournalEntryLine.objects.create(
            entry=entry,
            account=self.revenue,
            description='Credit line',
            credit=Decimal('90.00')
        )

        # Intentar contabilizar debería fallar
        with self.assertRaises(ValidationError):
            entry.post()

    def test_trial_balance(self):
        """Verifica la generación del balance de comprobación"""
        # Crear y contabilizar un asiento
        entry = JournalEntry.objects.create(
            date=date.today(),
            description='Test Entry',
            created_by=self.user
        )

        JournalEntryLine.objects.create(
            entry=entry,
            account=self.cash,
            description='Debit line',
            debit=Decimal('100.00')
        )

        JournalEntryLine.objects.create(
            entry=entry,
            account=self.revenue,
            description='Credit line',
            credit=Decimal('100.00')
        )

        entry.post()

        # Generar balance de comprobación
        trial_balance = TrialBalance.objects.create(
            date=date.today(),
            created_by=self.user
        )

        trial_balance.generate()

        # Verificar saldos
        self.assertEqual(
            trial_balance.lines.aggregate(Sum('debit'))['debit__sum'],
            trial_balance.lines.aggregate(Sum('credit'))['credit__sum']
        )

Ejemplo Real: Sistema de Transferencias entre Cuentas

Crear una Transferencia Bancaria

def create_bank_transfer(
    date,
    amount,
    from_account,
    to_account,
    description,
    user
):
    """
    Crea un asiento contable para una transferencia bancaria.
    """
    entry = JournalEntry.objects.create(
        date=date,
        description=description,
        created_by=user
    )

    # Crear línea de débito (cuenta destino)
    JournalEntryLine.objects.create(
        entry=entry,
        account=to_account,
        description=f"Transferencia recibida de {from_account.name}",
        debit=amount
    )

    # Crear línea de crédito (cuenta origen)
    JournalEntryLine.objects.create(
        entry=entry,
        account=from_account,
        description=f"Transferencia enviada a {to_account.name}",
        credit=amount
    )

    # Contabilizar el asiento
    entry.post()

    return entry

# Ejemplo de uso:
from decimal import Decimal
from datetime import date

transfer = create_bank_transfer(
    date=date.today(),
    amount=Decimal('1000.00'),
    from_account=Account.objects.get(code='1001'),  # Cuenta Banco
    to_account=Account.objects.get(code='1000'),    # Cuenta Caja
    description="Transferencia para gastos operativos",
    user=request.user
)

Mejores Prácticas

1. Validaciones de Seguridad

from django.contrib.admin import ModelAdmin
from django.core.exceptions import PermissionDenied

class JournalEntryAdmin(ModelAdmin):
    def has_delete_permission(self, request, obj=None):
        # Prevenir eliminación de asientos contabilizados
        if obj and obj.is_posted:
            return False
        return super().has_delete_permission(request, obj)

    def save_model(self, request, obj, form, change):
        # Verificar permisos especiales para contabilizar
        if 'is_posted' in form.changed_data:
            if not request.user.has_perm('accounting.post_journal_entry'):
                raise PermissionDenied("No tienes permiso para contabilizar asientos")
        super().save_model(request, obj, form, change)

2. Manejo de Errores

class AccountingError(Exception):
    """Base exception for accounting errors"""
    pass

class UnbalancedEntryError(AccountingError):
    """Raised when a journal entry is not balanced"""
    pass

class PostedEntryError(AccountingError):
    """Raised when trying to modify a posted entry"""
    pass

class JournalEntry(models.Model):
    # ... otros campos ...

    def post(self):
        try:
            # Validar balance
            total_debit = self.lines.aggregate(Sum('debit'))['debit__sum'] or 0
            total_credit = self.lines.aggregate(Sum('credit'))['credit__sum'] or 0

            if total_debit != total_credit:
                raise UnbalancedEntryError(
                    f"Débito ({total_debit}) != Crédito ({total_credit})"
                )

            # Validar que no esté contabilizado
            if self.is_posted:
                raise PostedEntryError("El asiento ya está contabilizado")

            # Contabilizar
            self.is_posted = True
            self.save()

        except AccountingError as e:
            # Log del error
            logger.error(f"Error al contabilizar asiento {self.reference}: {str(e)}")
            raise

3. Patrones de Diseño Recomendados

Command Pattern para Operaciones Contables

from abc import ABC, abstractmethod
from decimal import Decimal
from typing import List

class AccountingCommand(ABC):
    @abstractmethod
    def execute(self) -> JournalEntry:
        pass

class TransferCommand(AccountingCommand):
    def __init__(
        self,
        date: date,
        amount: Decimal,
        from_account: Account,
        to_account: Account,
        description: str,
        user: User
    ):
        self.date = date
        self.amount = amount
        self.from_account = from_account
        self.to_account = to_account
        self.description = description
        self.user = user

    def execute(self) -> JournalEntry:
        return create_bank_transfer(
            self.date,
            self.amount,
            self.from_account,
            self.to_account,
            self.description,
            self.user
        )

# Uso del Command Pattern
transfer_cmd = TransferCommand(
    date=date.today(),
    amount=Decimal('1000.00'),
    from_account=bank_account,
    to_account=cash_account,
    description="Transferencia operativa",
    user=current_user
)

journal_entry = transfer_cmd.execute()

Queries Útiles

1. Balance por Tipo de Cuenta

from django.db.models import Sum, Case, When, F

def get_account_type_balances(end_date=None):
    """
    Obtiene los saldos agrupados por tipo de cuenta
    """
    query = Account.objects.all()

    if end_date:
        lines = JournalEntryLine.objects.filter(
            entry__date__lte=end_date,
            entry__is_posted=True
        )
    else:
        lines = JournalEntryLine.objects.filter(entry__is_posted=True)

    balances = query.annotate(
        total_debit=Sum(
            Case(
                When(
                    journalentryline__in=lines,
                    then='journalentryline__debit'
                ),
                default=0
            )
        ),
        total_credit=Sum(
            Case(
                When(
                    journalentryline__in=lines,
                    then='journalentryline__credit'
                ),
                default=0
            )
        ),
        balance=Case(
            When(
                type__in=['ASSET', 'EXPENSE'],
                then=F('total_debit') - F('total_credit')
            ),
            default=F('total_credit') - F('total_debit')
        )
    ).values('type').annotate(
        total_balance=Sum('balance')
    )

    return balances

2. Libro Mayor

def get_ledger(account, start_date=None, end_date=None):
    """
    Obtiene el libro mayor para una cuenta específica
    """
    lines = JournalEntryLine.objects.filter(
        account=account,
        entry__is_posted=True
    ).select_related('entry')

    if start_date:
        lines = lines.filter(entry__date__gte=start_date)
    if end_date:
        lines = lines.filter(entry__date__lte=end_date)

    return lines.order_by('entry__date', 'entry__id')

Conclusión

En este tutorial, hemos construido un sistema contable robusto utilizando Django Admin, aplicando principios de programación familiares para entender conceptos contables. Los puntos clave son:

La contabilidad es similar a un sistema de logging transaccional
Las validaciones son cruciales para mantener la integridad de los datos
El patrón Command nos ayuda a encapsular operaciones contables complejas

Contabilidad para Django Developers: Implementando Reglas de Negocio Contables

Enrique Lazo Bello — Sat, 09 Nov 2024 00:47:03 +0000

Introducción

¿Alguna vez te has preguntado por qué los contadores insisten tanto en que "el debe igual al haber"? Como desarrollador de Django, probablemente estés más familiarizado con transacciones ACID que con asientos contables. Pero aquí está la sorpresa: los principios contables son muy similares a los conceptos de programación que ya conoces.

En este tutorial, transformaremos conceptos contables en código Django limpio y mantenible. Aprenderás a implementar las reglas de negocio básicas de contabilidad utilizando solo el admin de Django, sin necesidad de crear vistas personalizadas.

Al final de este tutorial, serás capaz de:

Implementar validaciones contables robustas
Garantizar la integridad de los datos financieros
Crear un sistema contable básico pero profesional

Prerrequisitos

# Crear un nuevo ambiente virtual
python -m venv env
source env/bin/activate  # En Windows: env\Scripts\activate

# Instalar dependencias
pip install django==5.0
pip install python-dateutil

Conceptos Clave

1. Asientos Contables = Transacciones Atómicas

Así como una transacción en base de datos debe ser atómica, un asiento contable debe cuadrar (debe = haber). No existe un estado intermedio válido.

2. Período Contable = Ventana de Tiempo Válida

Similar a un rate limiter o una ventana de caché, un período contable define cuándo se pueden realizar operaciones.

3. Plan de Cuentas = Enum Jerárquico

Las cuentas contables son como un enum jerárquico con estados (activa/inactiva) y metadatos.

Implementación Práctica

Primero, definamos nuestros modelos:

# accounting/models.py
from django.db import models
from django.core.exceptions import ValidationError
from django.utils import timezone
from dateutil.relativedelta import relativedelta
import decimal

class AccountingPeriod(models.Model):
    name = models.CharField(max_length=100)
    start_date = models.DateField()
    end_date = models.DateField()
    is_closed = models.BooleanField(default=False)

    class Meta:
        ordering = ['-start_date']

    def __str__(self):
        return f"{self.name} ({self.start_date} - {self.end_date})"

    def clean(self):
        if self.start_date > self.end_date:
            raise ValidationError("La fecha de inicio debe ser anterior a la fecha final")

        # Verificar solapamiento de períodos
        overlapping = AccountingPeriod.objects.filter(
            start_date__lte=self.end_date,
            end_date__gte=self.start_date
        ).exclude(pk=self.pk)

        if overlapping.exists():
            raise ValidationError("El período se solapa con otros existentes")

    def is_date_valid(self, date):
        return self.start_date <= date <= self.end_date and not self.is_closed

class Currency(models.Model):
    code = models.CharField(max_length=3, unique=True)
    name = models.CharField(max_length=100)
    symbol = models.CharField(max_length=5)
    is_active = models.BooleanField(default=True)

    def __str__(self):
        return f"{self.code} - {self.name}"

    class Meta:
        verbose_name_plural = "Currencies"

class ExchangeRate(models.Model):
    currency = models.ForeignKey(Currency, on_delete=models.PROTECT)
    date = models.DateField()
    rate = models.DecimalField(max_digits=10, decimal_places=4)

    class Meta:
        unique_together = ['currency', 'date']

    def __str__(self):
        return f"{self.currency.code} - {self.date} - {self.rate}"

    def clean(self):
        if self.rate <= 0:
            raise ValidationError("El tipo de cambio debe ser mayor a 0")

class Account(models.Model):
    ACCOUNT_TYPES = [
        ('A', 'Asset'),
        ('L', 'Liability'),
        ('E', 'Equity'),
        ('I', 'Income'),
        ('X', 'Expense'),
    ]

    code = models.CharField(max_length=20, unique=True)
    name = models.CharField(max_length=200)
    type = models.CharField(max_length=1, choices=ACCOUNT_TYPES)
    is_active = models.BooleanField(default=True)
    allow_movements = models.BooleanField(default=True)
    parent = models.ForeignKey(
        'self',
        null=True,
        blank=True,
        on_delete=models.PROTECT,
        related_name='children'
    )

    def __str__(self):
        return f"{self.code} - {self.name}"

    def clean(self):
        if self.parent:
            if self.parent.type != self.type:
                raise ValidationError("La cuenta padre debe ser del mismo tipo")
            if not self.code.startswith(self.parent.code):
                raise ValidationError("El código debe empezar con el código de la cuenta padre")

class JournalEntry(models.Model):
    date = models.DateField()
    number = models.CharField(max_length=20, unique=True)
    description = models.TextField()
    currency = models.ForeignKey(Currency, on_delete=models.PROTECT)
    is_posted = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return f"{self.number} - {self.date}"

    def clean(self):
        if not self.pk:  # Solo para nuevos registros
            # Generar número automático
            last_entry = JournalEntry.objects.order_by('-number').first()
            if last_entry:
                last_num = int(last_entry.number)
                self.number = str(last_num + 1).zfill(8)
            else:
                self.number = "00000001"

        # Verificar período contable
        period = AccountingPeriod.objects.filter(
            start_date__lte=self.date,
            end_date__gte=self.date,
            is_closed=False
        ).first()

        if not period:
            raise ValidationError("La fecha no corresponde a un período contable válido y abierto")

        # Verificar tipo de cambio
        if not ExchangeRate.objects.filter(
            currency=self.currency,
            date=self.date
        ).exists():
            raise ValidationError(
                f"No existe tipo de cambio para {self.currency.code} en la fecha {self.date}"
            )

    def save(self, *args, **kwargs):
        if not self.is_posted:
            self.verify_balance()
        super().save(*args, **kwargs)

    def verify_balance(self):
        """Verifica que el asiento cuadre (debe = haber)"""
        if self.pk:  # Solo para entradas existentes
            debit_sum = self.lines.aggregate(
                total=models.Sum('debit_amount')
            )['total'] or 0
            credit_sum = self.lines.aggregate(
                total=models.Sum('credit_amount')
            )['total'] or 0

            if debit_sum != credit_sum:
                raise ValidationError(
                    f"El asiento no cuadra. Debe: {debit_sum}, Haber: {credit_sum}"
                )

class JournalEntryLine(models.Model):
    entry = models.ForeignKey(
        JournalEntry,
        on_delete=models.CASCADE,
        related_name='lines'
    )
    account = models.ForeignKey(Account, on_delete=models.PROTECT)
    description = models.CharField(max_length=200)
    debit_amount = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=0
    )
    credit_amount = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=0
    )

    def clean(self):
        if not self.account.is_active:
            raise ValidationError("La cuenta no está activa")

        if not self.account.allow_movements:
            raise ValidationError("La cuenta no permite movimientos")

        if self.debit_amount < 0 or self.credit_amount < 0:
            raise ValidationError("Los montos no pueden ser negativos")

        if (self.debit_amount > 0 and self.credit_amount > 0):
            raise ValidationError("Una línea no puede tener debe y haber")

        if (self.debit_amount == 0 and self.credit_amount == 0):
            raise ValidationError("El monto debe ser mayor a 0")

Ahora, configuremos el admin de Django para manejar estos modelos:

# accounting/admin.py
from django.contrib import admin
from django.db.models import Sum
from .models import (
    AccountingPeriod,
    Currency,
    ExchangeRate,
    Account,
    JournalEntry,
    JournalEntryLine
)

@admin.register(AccountingPeriod)
class AccountingPeriodAdmin(admin.ModelAdmin):
    list_display = ['name', 'start_date', 'end_date', 'is_closed']
    search_fields = ['name']
    list_filter = ['is_closed']

@admin.register(Currency)
class CurrencyAdmin(admin.ModelAdmin):
    list_display = ['code', 'name', 'symbol', 'is_active']
    search_fields = ['code', 'name']
    list_filter = ['is_active']

@admin.register(ExchangeRate)
class ExchangeRateAdmin(admin.ModelAdmin):
    list_display = ['currency', 'date', 'rate']
    search_fields = ['currency__code']
    list_filter = ['currency', 'date']

@admin.register(Account)
class AccountAdmin(admin.ModelAdmin):
    list_display = [
        'code',
        'name',
        'type',
        'is_active',
        'allow_movements',
        'parent'
    ]
    search_fields = ['code', 'name']
    list_filter = ['type', 'is_active', 'allow_movements']
    ordering = ['code']

class JournalEntryLineInline(admin.TabularInline):
    model = JournalEntryLine
    extra = 2

    def get_readonly_fields(self, request, obj=None):
        if obj and obj.is_posted:
            return ['account', 'description', 'debit_amount', 'credit_amount']
        return []

@admin.register(JournalEntry)
class JournalEntryAdmin(admin.ModelAdmin):
    list_display = [
        'number',
        'date',
        'description',
        'currency',
        'is_posted',
        'total_amount'
    ]
    search_fields = ['number', 'description']
    list_filter = ['date', 'currency', 'is_posted']
    inlines = [JournalEntryLineInline]
    readonly_fields = ['number']

    def get_readonly_fields(self, request, obj=None):
        if obj and obj.is_posted:
            return ['date', 'number', 'description', 'currency', 'is_posted']
        return ['number']

    def total_amount(self, obj):
        if obj.pk:
            return obj.lines.aggregate(
                total=Sum('debit_amount')
            )['total'] or 0
        return 0

    def save_model(self, request, obj, form, change):
        if not change:  # Nuevo registro
            obj.save()
        else:
            if 'is_posted' in form.changed_data and obj.is_posted:
                # Verificar balance antes de marcar como contabilizado
                obj.verify_balance()
            obj.save()

Ejemplo Real: Sistema de Transferencias entre Cuentas

Veamos cómo implementar una transferencia entre cuentas bancarias:

# accounting/tests.py
from django.test import TestCase
from django.core.exceptions import ValidationError
from decimal import Decimal
from datetime import date
from .models import (
    AccountingPeriod,
    Currency,
    ExchangeRate,
    Account,
    JournalEntry,
    JournalEntryLine
)

class AccountingTests(TestCase):
    def setUp(self):
        # Crear período contable
        self.period = AccountingPeriod.objects.create(
            name="2024-01",
            start_date=date(2024, 1, 1),
            end_date=date(2024, 1, 31)
        )

        # Crear moneda y tipo de cambio
        self.currency = Currency.objects.create(
            code="USD",
            name="US Dollar",
            symbol="$"
        )

        self.exchange_rate = ExchangeRate.objects.create(
            currency=self.currency,
            date=date(2024, 1, 15),
            rate=Decimal('1.0')
        )

        # Crear cuentas bancarias
        self.bank_parent = Account.objects.create(
            code="1001",
            name="Bancos",
            type='A'
        )

        self.bank_account_1 = Account.objects.create(
            code="1001001",
            name="Banco 1",
            type='A',
            parent=self.bank_parent
        )

        self.bank_account_2 = Account.objects.create(
            code="1001002",
            name="Banco 2",
            type='A',
            parent=self.bank_parent
        )

    def test_bank_transfer(self):
        # Crear asiento de transferencia
        entry = JournalEntry.objects.create(
            date=date(2024, 1, 15),
            description="Transferencia entre bancos",
            currency=self.currency
        )

        # Débito al banco 2
        JournalEntryLine.objects.create(
            entry=entry,
            account=self.bank_account_2,
            description="Transferencia recibida",
            debit_amount=Decimal('1000.00')
        )

        # Crédito al banco 1
        JournalEntryLine.objects.create(
            entry=entry,
            account=self.bank_account_1,
            description="Transferencia enviada",
            credit_amount=Decimal('1000.00')
        )

        # Verificar balance
        entry.verify_balance()

        # Marcar como contabilizado
        entry.is_posted = True
        entry.save()


    def test_invalid_transfer(self):
        # Probar transferencia con montos desbalanceados
        entry = JournalEntry.objects.create(
            date=date(2024, 1, 15),
            description="Transferencia inválida",
            currency=self.currency
        )

        JournalEntryLine.objects.create(
            entry=entry,
            account=self.bank_account_2,
            description="Transferencia recibida",
            debit_amount=Decimal('1000.00')
        )

        JournalEntryLine.objects.create(
            entry=entry,
            account=self.bank_account_1,
            description="Transferencia enviada",
            credit_amount=Decimal('900.00')  # Monto diferente
        )

        # Debería levantar una excepción al verificar el balance
        with self.assertRaises(ValidationError):
            entry.verify_balance()

    def test_closed_period(self):
        # Cerrar el período
        self.period.is_closed = True
        self.period.save()

        # Intentar crear un asiento en período cerrado
        with self.assertRaises(ValidationError):
            JournalEntry.objects.create(
                date=date(2024, 1, 15),
                description="Asiento en período cerrado",
                currency=self.currency
            )

    def test_inactive_account(self):
        # Desactivar cuenta
        self.bank_account_1.is_active = False
        self.bank_account_1.save()

        entry = JournalEntry.objects.create(
            date=date(2024, 1, 15),
            description="Transferencia con cuenta inactiva",
            currency=self.currency
        )

        # Intentar crear línea con cuenta inactiva
        with self.assertRaises(ValidationError):
            JournalEntryLine.objects.create(
                entry=entry,
                account=self.bank_account_1,
                description="Débito a cuenta inactiva",
                debit_amount=Decimal('1000.00')
            )

Mejores Prácticas

1. Validaciones de Seguridad

Para garantizar la integridad de los datos contables, hemos implementado múltiples niveles de validación:

# accounting/models.py

class JournalEntry(models.Model):
    # ... (código anterior) ...

    def save(self, *args, **kwargs):
        if self.is_posted:
            raise ValidationError(
                "No se pueden modificar asientos contabilizados"
            )
        super().save(*args, **kwargs)

class JournalEntryLine(models.Model):
    # ... (código anterior) ...

    def save(self, *args, **kwargs):
        if self.entry.is_posted:
            raise ValidationError(
                "No se pueden modificar líneas de asientos contabilizados"
            )
        super().save(*args, **kwargs)

2. Manejo de Errores

Implementamos un middleware personalizado para manejar errores contables de manera amigable:

# accounting/middleware.py
from django.core.exceptions import ValidationError
from django.contrib import messages
from django.shortcuts import redirect

class AccountingErrorMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        response = self.get_response(request)
        return response

    def process_exception(self, request, exception):
        if isinstance(exception, ValidationError):
            messages.error(
                request,
                f"Error de validación contable: {str(exception)}"
            )
            return redirect('admin:index')
        return None

3. Patrones de Diseño Recomendados

Unit of Work: Implementado a través de las transacciones de Django
Repository Pattern: Usando el ORM de Django
Value Objects: Para manejo de montos y tipos de cambio

# accounting/utils.py
from decimal import Decimal
from dataclasses import dataclass
from typing import Optional

@dataclass
class Money:
    amount: Decimal
    currency: str

    def __post_init__(self):
        self.amount = Decimal(str(self.amount))

    def __add__(self, other):
        if not isinstance(other, Money):
            raise TypeError("Solo se pueden sumar objetos Money")
        if self.currency != other.currency:
            raise ValueError("No se pueden sumar montos en diferentes monedas")
        return Money(self.amount + other.amount, self.currency)

    def __sub__(self, other):
        if not isinstance(other, Money):
            raise TypeError("Solo se pueden restar objetos Money")
        if self.currency != other.currency:
            raise ValueError("No se pueden restar montos en diferentes monedas")
        return Money(self.amount - other.amount, self.currency)

@dataclass
class ExchangeRateValue:
    from_currency: str
    to_currency: str
    rate: Decimal
    date: date

    def convert(self, money: Money) -> Money:
        if money.currency != self.from_currency:
            raise ValueError("Moneda incompatible")
        return Money(
            money.amount * self.rate,
            self.to_currency
        )

Consultas Útiles

Aquí hay algunos ejemplos de consultas comunes:

# Obtener balance por cuenta
from django.db.models import Sum, F, Value
from django.db.models.functions import Coalesce

def get_account_balance(account_code, date_from, date_to):
    return JournalEntryLine.objects.filter(
        account__code__startswith=account_code,
        entry__date__range=[date_from, date_to],
        entry__is_posted=True
    ).aggregate(
        balance=Coalesce(
            Sum(F('debit_amount') - F('credit_amount')),
            Value(0)
        )
    )['balance']

# Libro mayor
def get_ledger(account_code, date_from, date_to):
    return JournalEntryLine.objects.filter(
        account__code__startswith=account_code,
        entry__date__range=[date_from, date_to],
        entry__is_posted=True
    ).select_related(
        'entry', 'account'
    ).order_by('entry__date', 'entry__number')

# Balance de comprobación
def get_trial_balance(date_from, date_to):
    return Account.objects.filter(
        lines__entry__date__range=[date_from, date_to],
        lines__entry__is_posted=True
    ).annotate(
        debit_sum=Coalesce(
            Sum('lines__debit_amount'),
            Value(0)
        ),
        credit_sum=Coalesce(
            Sum('lines__credit_amount'),
            Value(0)
        ),
        balance=F('debit_sum') - F('credit_sum')
    ).filter(balance__ne=0)

Conclusión

Hemos construido un sistema contable básico pero robusto utilizando Django. Los puntos clave a recordar son:

Las reglas contables se implementan como validaciones en los modelos
El admin de Django proporciona una interfaz lista para usar
Los tests son cruciales para garantizar la integridad contable

Contabilidad para Django Developers: Control de Cambios y Tipo de Cambio

Enrique Lazo Bello — Sat, 09 Nov 2024 00:36:22 +0000

Introducción

Como desarrollador Django, ¿alguna vez te has preguntado cómo manejar correctamente las operaciones en diferentes monedas en tus aplicaciones? Si estás construyendo sistemas empresariales o financieros, entender el control de cambios no es solo una necesidad contable - es un requisito técnico crucial.

En este tutorial, transformaremos conceptos contables complejos en modelos Django intuitivos, usando el admin de Django como nuestra interfaz principal. Aprenderás a manejar tipos de cambio, calcular diferencias cambiarias y mantener una posición de cambio precisa, todo sin escribir una sola vista personalizada.

Al finalizar, tendrás un sistema robusto de control de cambios que cualquier contador podrá usar desde el admin de Django, con validaciones automáticas y tests que aseguran su correcto funcionamiento.

Prerrequisitos

Python 3.12+
Django 5.0+
Conocimientos básicos de modelos Django y el admin
SQLite o PostgreSQL
Git para control de versiones

Conceptos Clave

El Tipo de Cambio como un State Pattern

Piensa en el tipo de cambio como un state pattern en programación: representa el estado de una moneda en relación a otra en un momento específico. Así como guardamos timestamps para auditoría, necesitamos mantener un historial de tipos de cambio para operaciones financieras.

# Analogía en código
class Currency(str, Enum):
    PEN = "PEN"  # Soles
    USD = "USD"  # Dólares
    EUR = "EUR"  # Euros

# El tipo de cambio es como un estado inmutable
@dataclass(frozen=True)
class ExchangeRate:
    currency_from: Currency
    currency_to: Currency
    rate: Decimal
    timestamp: datetime

Diferencia de Cambio como Debugging

La diferencia de cambio es similar a encontrar un bug entre el comportamiento esperado y el actual. Cuando tienes una cuenta por pagar de $1000 registrada cuando el dólar estaba a S/3.80, pero la pagas cuando está a S/3.85, esa diferencia necesita ser registrada - como cuando haces un diff entre versiones de código.

Implementación Práctica

Modelos Base

from django.db import models
from django.core.validators import MinValueValidator
from decimal import Decimal
from django.core.exceptions import ValidationError
from django.utils import timezone

class Currency(models.Model):
    code = models.CharField(max_length=3, unique=True)
    name = models.CharField(max_length=50)
    is_base = models.BooleanField(default=False)

    def save(self, *args, **kwargs):
        if self.is_base:
            # Solo puede haber una moneda base
            Currency.objects.filter(is_base=True).update(is_base=False)
        super().save(*args, **kwargs)

    def __str__(self):
        return f"{self.code} - {self.name}"

    class Meta:
        verbose_name_plural = "Currencies"

class ExchangeRate(models.Model):
    currency_from = models.ForeignKey(
        Currency, 
        on_delete=models.PROTECT,
        related_name='rates_from'
    )
    currency_to = models.ForeignKey(
        Currency,
        on_delete=models.PROTECT,
        related_name='rates_to'
    )
    rate = models.DecimalField(
        max_digits=10,
        decimal_places=4,
        validators=[MinValueValidator(Decimal('0.0001'))]
    )
    date = models.DateField(default=timezone.now)
    is_sbs = models.BooleanField(
        default=False,
        verbose_name="Is SBS Rate",
        help_text="Indica si es tipo de cambio oficial SBS"
    )

    class Meta:
        unique_together = [['currency_from', 'currency_to', 'date']]

    def clean(self):
        if self.currency_from == self.currency_to:
            raise ValidationError(
                "La moneda origen y destino no pueden ser iguales"
            )
        if self.is_sbs and ExchangeRate.objects.filter(
            date=self.date,
            is_sbs=True,
            currency_from=self.currency_from,
            currency_to=self.currency_to
        ).exclude(id=self.id).exists():
            raise ValidationError(
                "Ya existe un tipo de cambio SBS para esta fecha"
            )

    def save(self, *args, **kwargs):
        self.full_clean()
        super().save(*args, **kwargs)

        # Crear tipo de cambio inverso automáticamente
        if not ExchangeRate.objects.filter(
            currency_from=self.currency_to,
            currency_to=self.currency_from,
            date=self.date
        ).exists():
            ExchangeRate.objects.create(
                currency_from=self.currency_to,
                currency_to=self.currency_from,
                rate=Decimal('1.0') / self.rate,
                date=self.date,
                is_sbs=self.is_sbs
            )

class ForexAccount(models.Model):
    currency = models.ForeignKey(Currency, on_delete=models.PROTECT)
    code = models.CharField(max_length=10, unique=True)
    name = models.CharField(max_length=100)
    balance = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=Decimal('0.00')
    )

    def __str__(self):
        return f"{self.code} - {self.name} ({self.currency.code})"

    def get_balance_base_currency(self, date=None):
        if not date:
            date = timezone.now().date()

        base_currency = Currency.objects.get(is_base=True)
        if self.currency == base_currency:
            return self.balance

        try:
            rate = ExchangeRate.objects.get(
                currency_from=self.currency,
                currency_to=base_currency,
                date=date
            )
            return self.balance * rate.rate
        except ExchangeRate.DoesNotExist:
            raise ValidationError(
                f"No existe tipo de cambio para {date}"
            )

class ForexTransaction(models.Model):
    TRANSACTION_TYPES = [
        ('IN', 'Ingreso'),
        ('OUT', 'Egreso'),
    ]

    account = models.ForeignKey(
        ForexAccount,
        on_delete=models.PROTECT,
        related_name='transactions'
    )
    type = models.CharField(max_length=3, choices=TRANSACTION_TYPES)
    amount = models.DecimalField(max_digits=15, decimal_places=2)
    exchange_rate = models.ForeignKey(
        ExchangeRate,
        on_delete=models.PROTECT,
        null=True,
        blank=True
    )
    date = models.DateField(default=timezone.now)
    description = models.TextField()

    def clean(self):
        if self.exchange_rate and (
            self.exchange_rate.currency_from != self.account.currency
        ):
            raise ValidationError(
                "El tipo de cambio debe corresponder a la moneda de la cuenta"
            )

    def save(self, *args, **kwargs):
        self.full_clean()
        if not self.exchange_rate:
            base_currency = Currency.objects.get(is_base=True)
            self.exchange_rate = ExchangeRate.objects.get(
                currency_from=self.account.currency,
                currency_to=base_currency,
                date=self.date
            )

        # Actualizar saldo de la cuenta
        multiplier = Decimal('1.0') if self.type == 'IN' else Decimal('-1.0')
        self.account.balance += (self.amount * multiplier)
        self.account.save()

        super().save(*args, **kwargs)

Configuración del Admin

from django.contrib import admin
from django.utils.html import format_html

@admin.register(Currency)
class CurrencyAdmin(admin.ModelAdmin):
    list_display = ['code', 'name', 'is_base']
    search_fields = ['code', 'name']
    list_filter = ['is_base']

@admin.register(ExchangeRate)
class ExchangeRateAdmin(admin.ModelAdmin):
    list_display = [
        'date',
        'currency_from',
        'currency_to',
        'rate',
        'is_sbs'
    ]
    list_filter = ['date', 'currency_from', 'currency_to', 'is_sbs']
    search_fields = ['currency_from__code', 'currency_to__code']
    date_hierarchy = 'date'

@admin.register(ForexAccount)
class ForexAccountAdmin(admin.ModelAdmin):
    list_display = [
        'code',
        'name',
        'currency',
        'balance',
        'base_currency_balance'
    ]
    search_fields = ['code', 'name']
    list_filter = ['currency']

    def base_currency_balance(self, obj):
        try:
            base_balance = obj.get_balance_base_currency()
            base_currency = Currency.objects.get(is_base=True)
            return f"{base_currency.code} {base_balance:.2f}"
        except Exception as e:
            return format_html(
                '<span style="color: red;">{}</span>',
                str(e)
            )

    base_currency_balance.short_description = "Balance en Moneda Base"

@admin.register(ForexTransaction)
class ForexTransactionAdmin(admin.ModelAdmin):
    list_display = [
        'date',
        'account',
        'type',
        'amount',
        'exchange_rate',
        'amount_base_currency'
    ]
    list_filter = ['date', 'type', 'account__currency']
    search_fields = ['description', 'account__name']
    date_hierarchy = 'date'

    def amount_base_currency(self, obj):
        base_amount = obj.amount * obj.exchange_rate.rate
        base_currency = Currency.objects.get(is_base=True)
        return f"{base_currency.code} {base_amount:.2f}"

    amount_base_currency.short_description = "Monto en Moneda Base"

Tests Unitarios

from django.test import TestCase
from django.core.exceptions import ValidationError
from decimal import Decimal
from django.utils import timezone
from .models import Currency, ExchangeRate, ForexAccount, ForexTransaction

class ForexTests(TestCase):
    def setUp(self):
        # Crear monedas base
        self.pen = Currency.objects.create(
            code='PEN',
            name='Soles',
            is_base=True
        )
        self.usd = Currency.objects.create(
            code='USD',
            name='Dólares'
        )

        # Crear tipo de cambio
        self.exchange_rate = ExchangeRate.objects.create(
            currency_from=self.usd,
            currency_to=self.pen,
            rate=Decimal('3.85'),
            date=timezone.now().date(),
            is_sbs=True
        )

        # Crear cuenta
        self.account = ForexAccount.objects.create(
            currency=self.usd,
            code='1041',
            name='Cuenta Corriente USD'
        )

    def test_exchange_rate_inverse(self):
        """Verifica que se cree automáticamente el tipo de cambio inverso"""
        inverse_rate = ExchangeRate.objects.get(
            currency_from=self.pen,
            currency_to=self.usd,
            date=self.exchange_rate.date
        )
        self.assertAlmostEqual(
            float(inverse_rate.rate),
            float(Decimal('1.0') / self.exchange_rate.rate),
            places=4
        )

    def test_account_balance_update(self):
        """Verifica la actualización correcta del saldo"""
        initial_balance = self.account.balance

        # Crear transacción de ingreso
        transaction = ForexTransaction.objects.create(
            account=self.account,
            type='IN',
            amount=Decimal('1000.00'),
            exchange_rate=self.exchange_rate,
            description='Test transaction'
        )

        self.account.refresh_from_db()
        self.assertEqual(
            self.account.balance,
            initial_balance + Decimal('1000.00')
        )

        # Verificar balance en moneda base
        base_balance = self.account.get_balance_base_currency()
        self.assertEqual(
            base_balance,
            Decimal('1000.00') * Decimal('3.85')
        )

    def test_invalid_exchange_rate(self):
        """Verifica que no se pueden crear tipos de cambio inválidos"""
        with self.assertRaises(ValidationError):
            ExchangeRate.objects.create(
                currency_from=self.usd,
                currency_to=self.usd,
                rate=Decimal('1.00'),
                date=timezone.now().date()
            )

Ejemplo Real: Sistema de Transferencias

Caso de Uso: Transferencia entre Cuentas

from django.db import transaction
from decimal import Decimal

def transfer_between_accounts(
    from_account: ForexAccount,
    to_account: ForexAccount,
    amount: Decimal,
    date=None
) -> tuple[ForexTransaction, ForexTransaction]:
    """
    Realiza una transferencia entre cuentas, manejando la conversión
    de monedas si es necesario.
    """
    if not date:
        date = timezone.now().date()

    with transaction.atomic():
        # Obtener tipo de cambio si las monedas son diferentes
        exchange_rate = None
        if from_account.currency != to_account.currency:
            exchange_rate = ExchangeRate.objects.get(
                currency_from=from_account.currency,
                currency_to=to_account.currency,
                date=date
            )

        # Crear transacción de salida
        outgoing = ForexTransaction.objects.create(
            account=from_account,
            type='OUT',
            amount=amount,
            date=date,
            description=f'Transferencia a {to_account.code}'
        )

        # Calcular monto en moneda destino si es necesario
        incoming_amount = amount
        if exchange_rate:
            incoming_amount = amount * exchange_rate.rate

        # Crear transacción de ingreso
        incoming = ForexTransaction.objects.create(
            account=to_account,
            type='IN',
            amount=incoming_amount,
            date=date,
            description=f'Transferencia desde {from_account.code}'
        )

        return outgoing, incoming

Mejores Prácticas

Validaciones de Seguridad
- Usar decimal.Decimal para todos los cálculos monetarios

Mejores Prácticas (continuación)

Validaciones de Seguridad
- Usar decimal.Decimal para todos los cálculos monetarios
- Implementar validaciones a nivel de modelo
- Utilizar transacciones atómicas para operaciones críticas
- Mantener un registro de auditoría

# models.py
class AuditMixin(models.Model):
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    created_by = models.ForeignKey(
        'auth.User',
        on_delete=models.PROTECT,
        related_name='%(class)s_created',
        null=True
    )
    updated_by = models.ForeignKey(
        'auth.User',
        on_delete=models.PROTECT,
        related_name='%(class)s_updated',
        null=True
    )

    class Meta:
        abstract = True

class ForexTransaction(AuditMixin, models.Model):
    # ... (código anterior del modelo) ...

    def save(self, *args, **kwargs):
        if not self.pk:  # Solo en creación
            if not hasattr(self, 'created_by') and hasattr(self, '_current_user'):
                self.created_by = self._current_user
        if hasattr(self, '_current_user'):
            self.updated_by = self._current_user
        super().save(*args, **kwargs)

Manejo de Errores
- Implementar excepciones personalizadas
- Logging detallado
- Mensajes de error amigables en el admin

# exceptions.py
class ForexError(Exception):
    """Base exception for forex operations"""
    pass

class ExchangeRateError(ForexError):
    """Raised when there's an issue with exchange rates"""
    pass

class InsufficientFundsError(ForexError):
    """Raised when an account has insufficient funds"""
    pass

# models.py
from django.core.exceptions import ValidationError
import logging

logger = logging.getLogger(__name__)

class ForexAccount(AuditMixin, models.Model):
    def withdraw(self, amount: Decimal) -> None:
        if self.balance < amount:
            logger.error(
                f"Insufficient funds in account {self.code}: "
                f"balance={self.balance}, amount={amount}"
            )
            raise InsufficientFundsError(
                f"Saldo insuficiente. Disponible: {self.balance}"
            )
        self.balance -= amount
        self.save()

Patrones de Diseño Recomendados
- Repository Pattern para acceso a datos
- Service Layer para lógica de negocio
- Command Pattern para operaciones complejas

# services.py
class ForexService:
    @staticmethod
    def calculate_exchange_difference(
        account: ForexAccount,
        date_from: date,
        date_to: date
    ) -> Decimal:
        """
        Calcula la diferencia de cambio para una cuenta en un período
        """
        initial_rate = ExchangeRate.objects.get(
            currency_from=account.currency,
            currency_to=Currency.objects.get(is_base=True),
            date=date_from
        )

        final_rate = ExchangeRate.objects.get(
            currency_from=account.currency,
            currency_to=Currency.objects.get(is_base=True),
            date=date_to
        )

        balance = account.balance
        initial_value = balance * initial_rate.rate
        final_value = balance * final_rate.rate

        return final_value - initial_value

class ForexReportService:
    @staticmethod
    def get_position_report(date: date) -> dict:
        """
        Genera reporte de posición de cambio
        """
        base_currency = Currency.objects.get(is_base=True)
        accounts = ForexAccount.objects.all()

        position = {
            'date': date,
            'positions': [],
            'total_base_currency': Decimal('0.00')
        }

        for account in accounts:
            if account.currency != base_currency:
                balance_base = account.get_balance_base_currency(date)
                position['positions'].append({
                    'account': account.code,
                    'currency': account.currency.code,
                    'balance': account.balance,
                    'balance_base': balance_base
                })
                position['total_base_currency'] += balance_base

        return position

Tests Adicionales

# tests.py
class ForexServiceTests(TestCase):
    def setUp(self):
        # ... (setup anterior) ...

    def test_exchange_difference_calculation(self):
        """
        Verifica el cálculo correcto de la diferencia de cambio
        """
        # Crear tipos de cambio para diferentes fechas
        date1 = timezone.now().date()
        date2 = date1 + timedelta(days=1)

        ExchangeRate.objects.create(
            currency_from=self.usd,
            currency_to=self.pen,
            rate=Decimal('3.85'),
            date=date1
        )

        ExchangeRate.objects.create(
            currency_from=self.usd,
            currency_to=self.pen,
            rate=Decimal('3.90'),
            date=date2
        )

        # Establecer saldo en la cuenta
        self.account.balance = Decimal('1000.00')
        self.account.save()

        # Calcular diferencia de cambio
        difference = ForexService.calculate_exchange_difference(
            self.account,
            date1,
            date2
        )

        # La diferencia debería ser (3.90 - 3.85) * 1000 = 50
        self.assertEqual(difference, Decimal('50.00'))

    def test_position_report(self):
        """
        Verifica la generación correcta del reporte de posición
        """
        date = timezone.now().date()

        # Crear algunas cuentas con saldos
        ForexAccount.objects.create(
            currency=self.usd,
            code='1042',
            name='Otra cuenta USD',
            balance=Decimal('2000.00')
        )

        position = ForexReportService.get_position_report(date)

        self.assertEqual(len(position['positions']), 2)
        self.assertGreater(position['total_base_currency'], Decimal('0'))

Conclusión

Has aprendido a implementar un sistema robusto de control de cambios en Django que:

Maneja tipos de cambio y diferencias cambiarias
Implementa validaciones y seguridad
Proporciona reportes de posición de cambio
Es fácilmente extensible

Contabilidad para Django Developers: Implementando Asientos Contables

Enrique Lazo Bello — Sat, 09 Nov 2024 00:23:12 +0000

Introducción

Como desarrollador de Django, ¿alguna vez te has enfrentado al desafío de implementar un sistema contable y te has sentido abrumado por la terminología y los conceptos? No estás solo. La contabilidad puede parecer un mundo completamente diferente, pero en realidad tiene muchas similitudes con la programación.

Este tutorial te guiará paso a paso en la implementación de un sistema de asientos contables en Django, utilizando analogías familiares para desarrolladores. Aprenderás a manejar diferentes tipos de asientos contables (apertura, operativos, ajustes y cierre) de una manera que tenga sentido para tu mente de programador.

Al final de este tutorial, serás capaz de implementar un sistema contable robusto utilizando solo el admin de Django, con validaciones sólidas y pruebas automatizadas.

Prerrequisitos

Python 3.12
Django 5.0+
Conocimientos básicos de modelos en Django
Familiaridad con el admin de Django

Conceptos Clave

Asientos Contables = Transacciones en Base de Datos

Piensa en los asientos contables como transacciones en una base de datos:

Asiento de Apertura: Es como el initial migration de tus modelos
Asientos Operativos: Similar a los CRUD operations diarios
Asientos de Ajuste: Como un database cleanup script
Asientos de Cierre: Equivalente a un end-of-period backup

Implementación de Modelos

from django.db import models
from django.core.exceptions import ValidationError
from django.db.models import Sum
from decimal import Decimal
from enum import Enum
import datetime

class AsientoTipo(models.TextChoices):
    APERTURA = 'AP', 'Apertura'
    OPERATIVO = 'OP', 'Operativo'
    AJUSTE = 'AJ', 'Ajuste'
    CIERRE = 'CI', 'Cierre'

class Cuenta(models.Model):
    codigo = models.CharField(max_length=20, unique=True)
    nombre = models.CharField(max_length=100)
    es_deudora = models.BooleanField(
        help_text="True para cuentas que aumentan con débito"
    )

    def __str__(self):
        return f"{self.codigo} - {self.nombre}"

    class Meta:
        ordering = ['codigo']

class LibroContable(models.Model):
    año = models.PositiveIntegerField()
    esta_cerrado = models.BooleanField(default=False)

    class Meta:
        unique_together = ['año']

    def clean(self):
        if self.esta_cerrado:
            if Asiento.objects.filter(
                libro=self, 
                tipo=AsientoTipo.CIERRE
            ).count() == 0:
                raise ValidationError(
                    "No se puede cerrar un libro sin asiento de cierre"
                )

    def __str__(self):
        return f"Libro Contable {self.año}"

class Asiento(models.Model):
    libro = models.ForeignKey(LibroContable, on_delete=models.PROTECT)
    fecha = models.DateField()
    tipo = models.CharField(
        max_length=2,
        choices=AsientoTipo.choices,
        default=AsientoTipo.OPERATIVO
    )
    descripcion = models.TextField()

    def clean(self):
        self._validar_fecha()
        self._validar_tipo_asiento()
        self._validar_balance()

    def _validar_fecha(self):
        if self.fecha.year != self.libro.año:
            raise ValidationError(
                "La fecha debe corresponder al año del libro contable"
            )

    def _validar_tipo_asiento(self):
        if self.tipo == AsientoTipo.APERTURA:
            self._validar_apertura()
        elif self.tipo == AsientoTipo.CIERRE:
            self._validar_cierre()

    def _validar_apertura(self):
        if Asiento.objects.filter(
            libro=self.libro,
            tipo=AsientoTipo.APERTURA
        ).exclude(pk=self.pk).exists():
            raise ValidationError(
                "Ya existe un asiento de apertura para este libro"
            )

    def _validar_cierre(self):
        if Asiento.objects.filter(
            libro=self.libro,
            tipo=AsientoTipo.CIERRE
        ).exclude(pk=self.pk).exists():
            raise ValidationError(
                "Ya existe un asiento de cierre para este libro"
            )

    def _validar_balance(self):
        if hasattr(self, 'pk'):
            total_debe = Decimal('0')
            total_haber = Decimal('0')
            for detalle in self.detalles.all():
                total_debe += detalle.debe
                total_haber += detalle.haber
            if total_debe != total_haber:
                raise ValidationError(
                    "El asiento debe estar balanceado (debe = haber)"
                )

    def __str__(self):
        return f"{self.fecha} - {self.get_tipo_display()} - {self.descripcion[:50]}"

class DetalleAsiento(models.Model):
    asiento = models.ForeignKey(
        Asiento,
        related_name='detalles',
        on_delete=models.CASCADE
    )
    cuenta = models.ForeignKey(Cuenta, on_delete=models.PROTECT)
    debe = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=0
    )
    haber = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=0
    )

    def clean(self):
        if self.debe > 0 and self.haber > 0:
            raise ValidationError(
                "Un movimiento no puede tener debe y haber simultáneamente"
            )
        if self.debe == 0 and self.haber == 0:
            raise ValidationError(
                "El movimiento debe tener un valor en debe o haber"
            )

    class Meta:
        ordering = ['id']

Configuración del Admin

from django.contrib import admin
from django.db import models
from django.forms import ModelForm
from .models import LibroContable, Cuenta, Asiento, DetalleAsiento

class DetalleAsientoInline(admin.TabularInline):
    model = DetalleAsiento
    extra = 2
    min_num = 2

@admin.register(Asiento)
class AsientoAdmin(admin.ModelAdmin):
    list_display = ['fecha', 'tipo', 'descripcion', 'total_debe', 'total_haber']
    list_filter = ['tipo', 'libro']
    search_fields = ['descripcion']
    inlines = [DetalleAsientoInline]

    def total_debe(self, obj):
        return obj.detalles.aggregate(
            total=models.Sum('debe')
        )['total'] or 0

    def total_haber(self, obj):
        return obj.detalles.aggregate(
            total=models.Sum('haber')
        )['total'] or 0

@admin.register(Cuenta)
class CuentaAdmin(admin.ModelAdmin):
    list_display = ['codigo', 'nombre', 'es_deudora']
    search_fields = ['codigo', 'nombre']

@admin.register(LibroContable)
class LibroContableAdmin(admin.ModelAdmin):
    list_display = ['año', 'esta_cerrado']
    list_filter = ['esta_cerrado']

Tests Automatizados

from django.test import TestCase
from django.core.exceptions import ValidationError
from decimal import Decimal
from .models import LibroContable, Cuenta, Asiento, DetalleAsiento

class AsientoTests(TestCase):
    def setUp(self):
        self.libro = LibroContable.objects.create(año=2024)
        self.cuenta_caja = Cuenta.objects.create(
            codigo='1.1.1',
            nombre='Caja',
            es_deudora=True
        )
        self.cuenta_capital = Cuenta.objects.create(
            codigo='3.1.1',
            nombre='Capital',
            es_deudora=False
        )

    def test_asiento_apertura(self):
        asiento = Asiento.objects.create(
            libro=self.libro,
            fecha='2024-01-01',
            tipo=AsientoTipo.APERTURA,
            descripcion='Asiento de apertura'
        )

        DetalleAsiento.objects.create(
            asiento=asiento,
            cuenta=self.cuenta_caja,
            debe=Decimal('1000')
        )

        DetalleAsiento.objects.create(
            asiento=asiento,
            cuenta=self.cuenta_capital,
            haber=Decimal('1000')
        )

        asiento.full_clean()  # Debería validar sin errores

        # Intentar crear otro asiento de apertura
        with self.assertRaises(ValidationError):
            Asiento.objects.create(
                libro=self.libro,
                fecha='2024-01-01',
                tipo=AsientoTipo.APERTURA,
                descripcion='Segundo asiento de apertura'
            ).full_clean()

Ejemplo Real: Sistema de Transferencias

def realizar_transferencia(
    libro_id: int,
    fecha: str,
    cuenta_origen_id: int,
    cuenta_destino_id: int,
    monto: Decimal,
    descripcion: str
) -> Asiento:
    """
    Crea un asiento contable para una transferencia entre cuentas.
    """
    libro = LibroContable.objects.get(pk=libro_id)

    asiento = Asiento.objects.create(
        libro=libro,
        fecha=fecha,
        tipo=AsientoTipo.OPERATIVO,
        descripcion=descripcion
    )

    # Movimiento de salida
    DetalleAsiento.objects.create(
        asiento=asiento,
        cuenta_id=cuenta_origen_id,
        haber=monto
    )

    # Movimiento de entrada
    DetalleAsiento.objects.create(
        asiento=asiento,
        cuenta_id=cuenta_destino_id,
        debe=monto
    )

    asiento.full_clean()
    return asiento

Mejores Prácticas

Validaciones de Seguridad
- Implementar permisos por tipo de asiento
- Validar fechas y períodos contables
- Prevenir modificaciones en libros cerrados
Manejo de Errores
- Usar transacciones para operaciones atómicas
- Implementar rollback en caso de error
- Registrar auditoría de cambios
Patrones de Diseño
- Usar Factory Method para diferentes tipos de asientos
- Implementar Observer para notificaciones
- Aplicar Strategy para diferentes reglas contables

Conclusión

Has aprendido a implementar un sistema contable básico pero robusto en Django, utilizando solo el admin. Los conceptos clave que debes recordar son:

Los asientos contables son como transacciones en una base de datos
Las validaciones son cruciales para mantener la integridad de los datos
El admin de Django es suficiente para muchas operaciones contables

Contabilidad para Django Developers: Implementando Estados Financieros

Enrique Lazo Bello — Fri, 08 Nov 2024 23:49:12 +0000

Introducción

Como desarrollador Django, ¿alguna vez te has enfrentado al desafío de implementar un sistema contable y te has sentido abrumado por la terminología financiera? No estás solo. La contabilidad puede parecer un mundo completamente diferente a la programación, pero en realidad, comparte muchos conceptos similares a los que ya conoces.

En este tutorial, aprenderás a implementar un sistema de estados financieros utilizando Django y su potente admin interface, traduciendo conceptos contables a términos de programación que ya conoces. Por ejemplo, verás cómo un asiento contable es similar a una transacción en una base de datos, y cómo el principio de partida doble es comparable a mantener la integridad referencial.

Al final de este tutorial, serás capaz de implementar un sistema contable básico pero robusto, con validaciones automáticas y tests unitarios, todo desde el Django Admin.

Prerrequisitos

Python 3.12
Django 5.0
Conocimientos básicos de modelos Django y Django Admin
SQLite o PostgreSQL para la base de datos

Conceptos Clave

La Contabilidad desde una Perspectiva de Programador

Antes de sumergirnos en el código, vamos a establecer algunas analogías:

Balance General = Estado de la Aplicación
- Similar a cómo guardamos el estado de una aplicación
- Activo = Recursos (como RAM o espacio en disco)
- Pasivo = Recursos comprometidos (como procesos en background)
- Patrimonio = Recursos netos disponibles
Estado de Resultados = Logs de Operaciones
- Como un log de transacciones
- Ingresos = Datos entrantes
- Gastos = Recursos consumidos
- Utilidad = Resultado final del proceso

Implementación

Modelos Base

from django.db import models
from django.core.exceptions import ValidationError
from django.db.models import Sum
from decimal import Decimal
import uuid

class AccountType(models.TextChoices):
    ASSET = 'ASSET', 'Activo'
    LIABILITY = 'LIABILITY', 'Pasivo'
    EQUITY = 'EQUITY', 'Patrimonio'
    INCOME = 'INCOME', 'Ingreso'
    EXPENSE = 'EXPENSE', 'Gasto'

class Account(models.Model):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    code = models.CharField(max_length=20, unique=True)
    name = models.CharField(max_length=100)
    type = models.CharField(max_length=10, choices=AccountType.choices)
    balance = models.DecimalField(max_digits=15, decimal_places=2, default=0)
    parent = models.ForeignKey(
        'self', 
        null=True, 
        blank=True, 
        on_delete=models.PROTECT,
        related_name='children'
    )
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        ordering = ['code']
        indexes = [
            models.Index(fields=['code']),
            models.Index(fields=['type']),
        ]

    def __str__(self):
        return f"{self.code} - {self.name}"

    def clean(self):
        if self.parent and self.parent.type != self.type:
            raise ValidationError('La cuenta padre debe ser del mismo tipo.')

    def save(self, *args, **kwargs):
        self.full_clean()
        super().save(*args, **kwargs)

    @property
    def calculated_balance(self):
        """Calcula el balance incluyendo subcuentas"""
        balance = self.balance
        for child in self.children.all():
            balance += child.calculated_balance
        return balance

class JournalEntry(models.Model):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    date = models.DateField()
    description = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        verbose_name = 'Asiento Contable'
        verbose_name_plural = 'Asientos Contables'
        ordering = ['-date', '-created_at']

    def __str__(self):
        return f"Asiento #{self.id} - {self.date}"

    def clean(self):
        if hasattr(self, 'entries'):
            debits = sum(entry.debit for entry in self.entries.all())
            credits = sum(entry.credit for entry in self.entries.all())
            if debits != credits:
                raise ValidationError(
                    'Los débitos y créditos deben ser iguales.'
                )

    def save(self, *args, **kwargs):
        self.full_clean()
        super().save(*args, **kwargs)

class JournalEntryLine(models.Model):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    entry = models.ForeignKey(
        JournalEntry,
        on_delete=models.CASCADE,
        related_name='entries'
    )
    account = models.ForeignKey(
        Account,
        on_delete=models.PROTECT,
        related_name='entries'
    )
    debit = models.DecimalField(max_digits=15, decimal_places=2, default=0)
    credit = models.DecimalField(max_digits=15, decimal_places=2, default=0)
    description = models.CharField(max_length=200, blank=True)

    class Meta:
        verbose_name = 'Línea de Asiento'
        verbose_name_plural = 'Líneas de Asiento'

    def clean(self):
        if self.debit < 0 or self.credit < 0:
            raise ValidationError(
                'Los valores no pueden ser negativos.'
            )
        if self.debit > 0 and self.credit > 0:
            raise ValidationError(
                'Una línea no puede tener débito y crédito simultáneamente.'
            )
        if self.debit == 0 and self.credit == 0:
            raise ValidationError(
                'Debe especificar un débito o un crédito.'
            )

    def save(self, *args, **kwargs):
        self.full_clean()

        # Actualizar balance de la cuenta
        if self._state.adding:  # Solo para nuevos registros
            if self.debit > 0:
                self.account.balance += self.debit
            else:
                self.account.balance -= self.credit
            self.account.save()

        super().save(*args, **kwargs)

Configuración del Admin

from django.contrib import admin
from django.db import models
from .models import Account, JournalEntry, JournalEntryLine

class AccountAdmin(admin.ModelAdmin):
    list_display = ['code', 'name', 'type', 'balance', 'parent']
    list_filter = ['type']
    search_fields = ['code', 'name']
    readonly_fields = ['balance']

class JournalEntryLineInline(admin.TabularInline):
    model = JournalEntryLine
    extra = 2
    fields = ['account', 'debit', 'credit', 'description']

class JournalEntryAdmin(admin.ModelAdmin):
    list_display = ['id', 'date', 'description', 'total_debit', 'total_credit']
    inlines = [JournalEntryLineInline]

    def total_debit(self, obj):
        return sum(line.debit for line in obj.entries.all())

    def total_credit(self, obj):
        return sum(line.credit for line in obj.entries.all())

admin.site.register(Account, AccountAdmin)
admin.site.register(JournalEntry, JournalEntryAdmin)

Tests Unitarios

from django.test import TestCase
from django.core.exceptions import ValidationError
from decimal import Decimal
from .models import Account, JournalEntry, JournalEntryLine

class AccountingTests(TestCase):
    def setUp(self):
        # Crear cuentas de prueba
        self.cash = Account.objects.create(
            code='1000',
            name='Caja',
            type='ASSET'
        )
        self.bank = Account.objects.create(
            code='1001',
            name='Banco',
            type='ASSET'
        )
        self.capital = Account.objects.create(
            code='3000',
            name='Capital',
            type='EQUITY'
        )

    def test_double_entry_principle(self):
        """Prueba el principio de partida doble"""
        entry = JournalEntry.objects.create(
            date='2024-01-01',
            description='Aporte inicial'
        )

        # Crear líneas de asiento
        JournalEntryLine.objects.create(
            entry=entry,
            account=self.cash,
            debit=Decimal('1000.00')
        )
        JournalEntryLine.objects.create(
            entry=entry,
            account=self.capital,
            credit=Decimal('1000.00')
        )

        # Verificar balances
        self.cash.refresh_from_db()
        self.capital.refresh_from_db()

        self.assertEqual(self.cash.balance, Decimal('1000.00'))
        self.assertEqual(self.capital.balance, Decimal('-1000.00'))

    def test_invalid_entry(self):
        """Prueba validaciones de asientos inválidos"""
        entry = JournalEntry.objects.create(
            date='2024-01-01',
            description='Asiento inválido'
        )

        with self.assertRaises(ValidationError):
            JournalEntryLine.objects.create(
                entry=entry,
                account=self.cash,
                debit=Decimal('-1000.00')  # Valor negativo
            )

Ejemplo Práctico: Sistema de Transferencias

1. Crear las Cuentas Base

Desde el admin de Django, crea las siguientes cuentas:

Activos
- Caja (1000)
- Banco (1001)
Pasivos
- Cuentas por Pagar (2000)
Patrimonio
- Capital (3000)
Ingresos
- Ventas (4000)
Gastos
- Gastos Operativos (5000)

2. Registrar una Transacción

Para registrar una venta en efectivo:

Crear nuevo JournalEntry
Fecha: La fecha de la transacción
Descripción: "Venta al contado"
Agregar líneas:
- Débito a Caja (1000) por $100
- Crédito a Ventas (4000) por $100

3. Verificar Balances

# Desde el shell de Django
from accounting.models import Account

# Verificar balance de caja
caja = Account.objects.get(code='1000')
print(f"Balance de caja: ${caja.balance}")

# Verificar balance de ventas
ventas = Account.objects.get(code='4000')
print(f"Balance de ventas: ${ventas.balance}")

Mejores Prácticas

Validaciones de Seguridad
- Usar transacciones de base de datos
- Implementar permisos granulares en el admin
- Mantener log de cambios
Manejo de Errores
- Validar balances negativos según tipo de cuenta
- Verificar referencias circulares en cuentas
- Implementar rollback en caso de error
Patrones de Diseño
- Usar Factory Method para crear asientos predefinidos
- Implementar Observer para actualizar balances
- Aplicar Command para operaciones complejas

Conclusión

Has aprendido a implementar un sistema contable básico usando Django y su admin interface. Los conceptos clave que debes recordar son:

La contabilidad es similar a mantener estado en tu aplicación
El principio de partida doble garantiza la integridad de los datos
Las validaciones automáticas previenen errores comunes

Contabilidad para Django Developers: Períodos Contables

Enrique Lazo Bello — Fri, 08 Nov 2024 23:23:26 +0000

Introducción

Como desarrollador Django, probablemente has trabajado con sistemas que manejan datos temporales, pero los períodos contables son diferentes. Imagina un sistema donde necesitas garantizar que todas las operaciones financieras estén correctamente organizadas en períodos de tiempo específicos, y que una vez cerrado un período, los datos sean inmutables.

Este tutorial te guiará en la implementación de un sistema de períodos contables robusto que garantice la integridad de los datos financieros, utilizando únicamente el admin de Django. Aprenderás a manejar la apertura, operación y cierre de períodos contables, junto con todas las validaciones necesarias para mantener la integridad de los datos.

Al finalizar, serás capaz de implementar un sistema que cumpla con los principios básicos de contabilidad mientras mantienes las mejores prácticas de desarrollo en Django.

Prerrequisitos

Python 3.12
Django 5.0
Conocimientos básicos de modelos en Django
SQLite o PostgreSQL (recomendado)

Conceptos Clave

Período Contable: La Transacción Definitiva

Piensa en un período contable como una transacción de base de datos a gran escala:

La apertura es como iniciar una transacción (BEGIN TRANSACTION)
Las operaciones son como los queries dentro de la transacción
El cierre es como ejecutar el COMMIT

Estados del Período

Similar a los estados de un objeto en memoria:

DRAFT: Como un objeto instanciado pero no guardado
ACTIVE: Como un objeto persistido en la base de datos
CLOSED: Como un objeto inmutable

Implementación

Modelos Base

from django.db import models
from django.core.exceptions import ValidationError
from django.utils import timezone
from decimal import Decimal
from django.db.models import Sum
from datetime import datetime

class AccountingPeriod(models.Model):
    """
    Representa un período contable con su ciclo de vida completo.
    Similar a una transacción de base de datos a gran escala.
    """
    STATUS_CHOICES = [
        ('DRAFT', 'Borrador'),
        ('ACTIVE', 'Activo'),
        ('CLOSED', 'Cerrado'),
    ]

    name = models.CharField(
        max_length=100,
        unique=True,
        help_text="Ejemplo: 'Enero 2024'"
    )
    start_date = models.DateField()
    end_date = models.DateField()
    status = models.CharField(
        max_length=10,
        choices=STATUS_CHOICES,
        default='DRAFT'
    )
    initial_balance = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=Decimal('0.00')
    )
    created_at = models.DateTimeField(auto_now_add=True)
    closed_at = models.DateTimeField(null=True, blank=True)
    closing_notes = models.TextField(blank=True)

    class Meta:
        ordering = ['-start_date']
        constraints = [
            models.CheckConstraint(
                check=models.Q(end_date__gte=models.F('start_date')),
                name='valid_period_dates'
            )
        ]

    def clean(self):
        self._validate_dates()
        self._validate_status_transition()
        self._validate_overlapping()

    def _validate_dates(self):
        """Valida la coherencia de las fechas del período."""
        if self.start_date and self.end_date:
            if self.start_date > self.end_date:
                raise ValidationError(
                    'La fecha de inicio debe ser anterior a la fecha de fin.'
                )

    def _validate_status_transition(self):
        """Valida las transiciones de estado permitidas."""
        if not self.pk:
            return

        old_instance = AccountingPeriod.objects.get(pk=self.pk)
        valid_transitions = {
            'DRAFT': ['ACTIVE'],
            'ACTIVE': ['CLOSED'],
            'CLOSED': []
        }

        if (self.status != old_instance.status and 
            self.status not in valid_transitions[old_instance.status]):
            raise ValidationError(
                f'Transición de estado inválida: {old_instance.status} -> {self.status}'
            )

    def _validate_overlapping(self):
        """Previene la superposición de períodos contables."""
        overlapping = AccountingPeriod.objects.exclude(pk=self.pk).filter(
            models.Q(start_date__lte=self.end_date) & 
            models.Q(end_date__gte=self.start_date)
        )
        if overlapping.exists():
            raise ValidationError(
                'El período se superpone con períodos existentes.'
            )

    def activate(self):
        """Activa un período en estado DRAFT."""
        if self.status != 'DRAFT':
            raise ValidationError('Solo se pueden activar períodos en borrador.')

        self.status = 'ACTIVE'
        self.save()

    def can_close(self):
        """
        Verifica si el período puede ser cerrado.
        Retorna (bool, str) donde bool indica si se puede cerrar
        y str contiene el mensaje de error si no se puede.
        """
        if self.status != 'ACTIVE':
            return False, 'Solo se pueden cerrar períodos activos.'

        # Verificar balance
        if not self.is_balanced():
            return False, 'El período no está balanceado.'

        # Verificar entradas sin procesar
        if self.has_pending_entries():
            return False, 'Existen entradas pendientes de procesar.'

        return True, ''

    def close(self, closing_notes=''):
        """
        Cierra el período si todas las validaciones pasan.
        """
        can_close, message = self.can_close()
        if not can_close:
            raise ValidationError(message)

        self.status = 'CLOSED'
        self.closed_at = timezone.now()
        self.closing_notes = closing_notes
        self.save()

    def is_balanced(self):
        """
        Verifica si los débitos y créditos están balanceados.
        """
        totals = self.entries.aggregate(
            total_debit=Sum('debit_amount'),
            total_credit=Sum('credit_amount')
        )

        total_debit = totals['total_debit'] or Decimal('0.00')
        total_credit = totals['total_credit'] or Decimal('0.00')

        return abs(total_debit - total_credit) < Decimal('0.01')

    def has_pending_entries(self):
        """
        Verifica si hay entradas pendientes de procesar.
        """
        return self.entries.filter(status='PENDING').exists()

    def __str__(self):
        return f"{self.name} ({self.get_status_display()})"


class AccountingEntry(models.Model):
    """
    Representa una entrada contable individual dentro de un período.
    """
    STATUS_CHOICES = [
        ('PENDING', 'Pendiente'),
        ('PROCESSED', 'Procesado'),
        ('REJECTED', 'Rechazado'),
    ]

    period = models.ForeignKey(
        AccountingPeriod,
        on_delete=models.PROTECT,
        related_name='entries'
    )
    date = models.DateField()
    description = models.CharField(max_length=200)
    debit_amount = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=Decimal('0.00')
    )
    credit_amount = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=Decimal('0.00')
    )
    status = models.CharField(
        max_length=10,
        choices=STATUS_CHOICES,
        default='PENDING'
    )
    created_at = models.DateTimeField(auto_now_add=True)
    processed_at = models.DateTimeField(null=True, blank=True)

    class Meta:
        ordering = ['-date', '-created_at']

    def clean(self):
        self._validate_period()
        self._validate_amounts()
        self._validate_date()

    def _validate_period(self):
        """Valida que el período esté activo y la entrada pueda ser modificada."""
        if not self.period_id:
            self.period = self._get_appropriate_period()

        if self.period.status == 'CLOSED':
            raise ValidationError(
                'No se pueden crear o modificar entradas en períodos cerrados.'
            )

    def _validate_amounts(self):
        """Valida que los montos sean coherentes."""
        if self.debit_amount < 0 or self.credit_amount < 0:
            raise ValidationError(
                'Los montos no pueden ser negativos.'
            )

        if self.debit_amount == 0 and self.credit_amount == 0:
            raise ValidationError(
                'Al menos un monto debe ser mayor a cero.'
            )

    def _validate_date(self):
        """Valida que la fecha esté dentro del período."""
        if not (self.period.start_date <= self.date <= self.period.end_date):
            raise ValidationError(
                'La fecha debe estar dentro del período contable.'
            )

    def _get_appropriate_period(self):
        """Obtiene el período activo adecuado para la fecha de la entrada."""
        period = AccountingPeriod.objects.filter(
            start_date__lte=self.date,
            end_date__gte=self.date,
            status='ACTIVE'
        ).first()

        if not period:
            raise ValidationError(
                'No existe un período activo para la fecha especificada.'
            )
        return period

    def process(self):
        """Procesa la entrada contable."""
        if self.status != 'PENDING':
            raise ValidationError('Solo se pueden procesar entradas pendientes.')

        self.status = 'PROCESSED'
        self.processed_at = timezone.now()
        self.save()

    def reject(self):
        """Rechaza la entrada contable."""
        if self.status != 'PENDING':
            raise ValidationError('Solo se pueden rechazar entradas pendientes.')

        self.status = 'REJECTED'
        self.processed_at = timezone.now()
        self.save()

    def __str__(self):
        return f"{self.date} - {self.description}"

Configuración del Admin

from django.contrib import admin
from django.utils.html import format_html
from django.contrib import messages

@admin.register(AccountingPeriod)
class AccountingPeriodAdmin(admin.ModelAdmin):
    list_display = [
        'name', 
        'start_date', 
        'end_date', 
        'status_display', 
        'balance_display',
        'created_at'
    ]
    list_filter = ['status']
    search_fields = ['name']
    readonly_fields = ['closed_at']

    def status_display(self, obj):
        colors = {
            'DRAFT': '#6c757d',
            'ACTIVE': '#28a745',
            'CLOSED': '#dc3545',
        }
        return format_html(
            '<span style="color: {}; font-weight: bold;">{}</span>',
            colors[obj.status],
            obj.get_status_display()
        )
    status_display.short_description = 'Estado'

    def balance_display(self, obj):
        if obj.is_balanced():
            return format_html(
                '<span style="color: #28a745;">Balanceado</span>'
            )
        return format_html(
            '<span style="color: #dc3545;">Desbalanceado</span>'
        )
    balance_display.short_description = 'Balance'

    def get_readonly_fields(self, request, obj=None):
        if obj and obj.status == 'CLOSED':
            return [f.name for f in obj._meta.fields]
        return self.readonly_fields

    actions = ['activate_periods', 'close_periods']

    def activate_periods(self, request, queryset):
        for period in queryset:
            try:
                period.activate()
                self.message_user(
                    request,
                    f'Período {period.name} activado exitosamente.',
                    messages.SUCCESS
                )
            except ValidationError as e:
                self.message_user(
                    request,
                    f"Error al activar {period.name}: {str(e)}",
                    messages.ERROR
                )
    activate_periods.short_description = "Activar períodos seleccionados"

    def close_periods(self, request, queryset):
        for period in queryset:
            try:
                period.close()
                self.message_user(
                    request,
                    f'Período {period.name} cerrado exitosamente.',
                    messages.SUCCESS
                )
            except ValidationError as e:
                self.message_user(
                    request,
                    f"Error al cerrar {period.name}: {str(e)}",
                    messages.ERROR
                )
    close_periods.short_description = "Cerrar períodos seleccionados"

@admin.register(AccountingEntry)
class AccountingEntryAdmin(admin.ModelAdmin):
    list_display = [
        'date', 
        'description', 
        'debit_amount', 
        'credit_amount',
        'status_display', 
        'period'
    ]
    list_filter = ['period', 'status', 'date']
    search_fields = ['description']
    actions = ['process_entries', 'reject_entries']

    def status_display(self, obj):
        colors = {
            'PENDING': '#ffc107',
            'PROCESSED': '#28a745',
            'REJECTED': '#dc3545',
        }
        return format_html(
            '<span style="color: {}; font-weight: bold;">{}</span>',
            colors[obj.status],
            obj.get_status_display()
        )
    status_display.short_description = 'Estado'

    def get_readonly_fields(self, request, obj=None):
        if obj and (obj.status != 'PENDING' or obj.period.status == 'CLOSED'):
            return [f.name for f in obj._meta.fields]
        return []

    def has_delete_permission(self, request, obj=None):
        if obj and (obj.status != 'PENDING' or obj.period.status == 'CLOSED'):
            return False
        return super().has_delete_permission(request, obj)

    def process_entries(self, request, queryset):
        for entry in queryset:
            try:
                entry.process()
                self.message_user(
                    request,
                    f'Entrada {entry} procesada exitosamente.',
                    messages.SUCCESS
                )
            except ValidationError as e:
                self.message_user(
                    request,
                    f"Error al procesar entrada {entry}: {str(e)}",
                    messages.ERROR
                )
    process_entries.short_description = "Procesar entradas seleccionadas"

## Tests Unitarios

python
from django.test import TestCase
from django.core.exceptions import ValidationError
from decimal import Decimal
from datetime import date, timedelta
from .models import AccountingPeriod, AccountingEntry

class AccountingPeriodTests(TestCase):
def setUp(self):
self.period = AccountingPeriod.objects.create(
name="Test Period",
start_date=date(2024, 1, 1),
end_date=date(2024, 1, 31),
status='DRAFT'
)

def test_period_lifecycle(self):
    """Test del ciclo de vida completo de un período"""
    # Verificar estado inicial
    self.assertEqual(self.period.status, 'DRAFT')

    # Activar período
    self.period.activate()
    self.assertEqual(self.period.status, 'ACTIVE')

    # Crear entradas balanceadas
    AccountingEntry.objects.create(
        period=self.period,
        date=date(2024, 1, 15),
        description="Test Entry 1",
        debit_amount=Decimal('100.00'),
        credit_amount=Decimal('0.00'),
        status='PROCESSED'
    )
    AccountingEntry.objects.create(
        period=self.period,
        date=date(2024, 1, 15),
        description="Test Entry 2",
        debit_amount=Decimal('0.00'),
        credit_amount=Decimal('100.00'),
        status='PROCESSED'
    )

    # Cerrar período
    self.period.close()
    self.assertEqual(self.period.status, 'CLOSED')
    self.assertIsNotNone(self.period.closed_at)

def test_overlapping_periods(self):
    """Test de validación de períodos superpuestos"""
    with self.assertRaises(ValidationError):
        AccountingPeriod.objects.create(
            name="Overlapping Period",
            start_date=date(2024, 1, 15),
            end_date=date(2024, 2, 15),
            status='DRAFT'
        )

def test_invalid_status_transition(self):
    """Test de transiciones de estado inválidas"""
    # No se puede pasar directamente de DRAFT a CLOSED
    self.period.status = 'CLOSED'
    with self.assertRaises(ValidationError):
        self.period.save()

    # No se puede reabrir un período cerrado
    self.period.activate()
    self.period.close()
    self.period.status = 'ACTIVE'
    with self.assertRaises(ValidationError):
        self.period.save()

class AccountingEntryTests(TestCase):
def setUp(self):
self.period = AccountingPeriod.objects.create(
name="Test Period",
start_date=date(2024, 1, 1),
end_date=date(2024, 1, 31),
status='ACTIVE'
)

def test_entry_validation(self):
    """Test de validaciones básicas de entradas"""
    # Entrada con montos negativos
    with self.assertRaises(ValidationError):
        AccountingEntry.objects.create(
            period=self.period,
            date=date(2024, 1, 15),
            description="Invalid Entry",
            debit_amount=Decimal('-100.00'),
            credit_amount=Decimal('0.00')
        )

    # Entrada fuera del período
    with self.assertRaises(ValidationError):
        AccountingEntry.objects.create(
            period=self.period,
            date=date(2024, 2, 1),
            description="Invalid Entry",
            debit_amount=Decimal('100.00'),
            credit_amount=Decimal('0.00')
        )

def test_entry_processing(self):
    """Test del procesamiento de entradas"""
    entry = AccountingEntry.objects.create(
        period=self.period,
        date=date(2024, 1, 15),
        description="Test Entry",
        debit_amount=Decimal('100.00'),
        credit_amount=Decimal('0.00')
    )

    # Procesar entrada
    entry.process()
    self.assertEqual(entry.status, 'PROCESSED')
    self.assertIsNotNone(entry.processed_at)

    # No se puede procesar una entrada ya procesada
    with self.assertRaises(ValidationError):
        entry.process()

def test_closed_period_modifications(self):
    """Test de modificaciones en período cerrado"""
    self.period.close()

    # No se pueden crear nuevas entradas
    with self.assertRaises(ValidationError):
        AccountingEntry.objects.create(
            period=self.period,
            date=date(2024, 1, 15),
            description="Test Entry",
            debit_amount=Decimal('100.00'),
            credit_amount=Decimal('0.00')
        )


## Ejemplo Práctico: Sistema de Transferencias

python

Ejemplo de uso del sistema

from decimal import Decimal
from datetime import date

def transfer_between_accounts(
period_id: int,
date: date,
amount: Decimal,
description: str
) -> tuple:
"""
Realiza una transferencia entre cuentas creando dos entradas contables.

Returns:
    tuple: (debit_entry, credit_entry)
"""
try:
    period = AccountingPeriod.objects.get(id=period_id)

    # Crear entrada de débito
    debit_entry = AccountingEntry.objects.create(
        period=period,
        date=date,
        description=f"DÉBITO - {description}",
        debit_amount=amount,
        credit_amount=Decimal('0.00')
    )

    # Crear entrada de crédito
    credit_entry = AccountingEntry.objects.create(
        period=period,
        date=date,
        description=f"CRÉDITO - {description}",
        debit_amount=Decimal('0.00'),
        credit_amount=amount
    )

    # Procesar ambas entradas
    debit_entry.process()
    credit_entry.process()

    return debit_entry, credit_entry

except AccountingPeriod.DoesNotExist:
    raise ValidationError("Período contable no encontrado")
except Exception as e:
    raise ValidationError(f"Error en la transferencia: {str(e)}")

Ejemplo de uso:

try:
period = AccountingPeriod.objects.get(
start_date_lte=date(2024, 1, 15),
end_date_gte=date(2024, 1, 15),
status='ACTIVE'
)

debit, credit = transfer_between_accounts(
    period_id=period.id,
    date=date(2024, 1, 15),
    amount=Decimal('1000.00'),
    description="Transferencia entre cuentas"
)

print(f"Transferencia exitosa: {debit.id}, {credit.id}")

except ValidationError as e:
print(f"Error: {str(e)}")


## Queries Útiles

python

Verificar balance de un período

def check_period_balance(period_id: int) -> dict:
period = AccountingPeriod.objects.get(id=period_id)
totals = period.entries.aggregate(
total_debit=Sum('debit_amount'),
total_credit=Sum('credit_amount')
)

return {
    'period_name': period.name,
    'total_debit': totals['total_debit'] or Decimal('0.00'),
    'total_credit': totals['total_credit'] or Decimal('0.00'),
    'is_balanced': period.is_balanced()
}

Obtener entradas pendientes

def get_pending_entries(period_id: int) -> QuerySet:
return AccountingEntry.objects.filter(
period_id=period_id,
status='PENDING'
)

Obtener resumen de período

def get_period_summary(period_id: int) -> dict:
period = AccountingPeriod.objects.get(id=period_id)
entries_summary = period.entries.aggregate(
total_entries=Count('id'),
pending_entries=Count('id', filter=Q(status='PENDING')),
processed_entries=Count('id', filter=Q(status='PROCESSED')),
rejected_entries=Count('id', filter=Q(status='REJECTED')),
total_debit=Sum('debit_amount'),
total_credit=Sum('credit_amount')
)

return {
    'period_name': period.name,
    'status': period.get_status_display(),
    'duration': (period.end_date - period.start_date).days + 1,
    'total_entries': entries_summary['total_entries'],
    'pending_entries': entries_summary['pending_entries'],
    'processed_entries': entries_summary['processed_entries'],
    'rejected_entries': entries_summary['rejected_entries'],
    'total_debit': entries_summary['total_debit'] or Decimal('0.00'),
    'total_credit': entries_summary['total_credit'] or Decimal('0.00')
}




## Mejores Prácticas

1. **Validaciones de Seguridad**
   - Usar `PROTECT` en foreign keys para prevenir eliminación en cascada
   - Implementar validaciones a nivel de modelo
   - Mantener registro de auditoría de cambios importantes

2. **Manejo de Errores**
   - Usar excepciones específicas y mensajes claros
   - Implementar rollback en operaciones complejas
   - Registrar errores críticos en logs

3. **Patrones de Diseño**
   - State Pattern para estados de período y entradas
   - Factory Method para crear entradas relacionadas
   - Observer Pattern para auditoría y notificaciones

## Conclusión

Este sistema proporciona una base sólida para manejar períodos contables en Django, con:
- Validaciones robustas para mantener la integridad de los datos
- Flujo claro de estados para períodos y entradas
- Sistema extensible para diferentes tipos de operaciones contables

Contabilidad para Django Developers: Implementando un Plan de Cuentas

Enrique Lazo Bello — Fri, 08 Nov 2024 16:45:18 +0000

Contabilidad para Django Developers: Implementando un Plan de Cuentas Jerárquico

Introducción

Si has trabajado con bases de datos jerárquicas o estructuras de árbol en programación, entonces ya tienes la base mental para entender un Plan de Cuentas contable. Imagina un sistema de archivos: tienes directorios principales (Grupos), subdirectorios (Clases), y archivos (Cuentas), cada uno con sus propias características y restricciones.

En este tutorial, traduciremos conceptos contables a términos de programación, implementando un Plan de Cuentas robusto que maneje la jerarquía natural de la contabilidad: Grupo → Clase → Cuenta → Subcuenta. Aprenderás a crear un sistema que mantenga la integridad de los datos contables mientras preserva las relaciones jerárquicas entre cuentas.

Prerrequisitos

# Crear entorno virtual
python -m venv env
source env/bin/activate

# Instalar dependencias
pip install django==5.0

Conceptos Clave

La Jerarquía Contable como Estructura de Árbol

# Analogía con sistema de archivos
root/
  ├── activos/              # Grupo
  │   ├── corrientes/       # Clase
  │   │   ├── efectivo/     # Cuenta
  │   │   │   ├── caja      # Subcuenta
  │   │   │   └── bancos    # Subcuenta
  │   │   └── inventario
  │   └── fijos/
  ├── pasivos/
  └── patrimonio/

Implementación en Django

from decimal import Decimal, ROUND_HALF_UP
from django.db import models
from django.core.exceptions import ValidationError
from django.core.validators import MinValueValidator, MaxValueValidator

class AccountLevel(models.TextChoices):
    """Niveles jerárquicos del plan de cuentas"""
    GROUP = 'G', 'Grupo'
    CLASS = 'C', 'Clase'
    ACCOUNT = 'A', 'Cuenta'
    SUBACCOUNT = 'S', 'Subcuenta'

class AccountType(models.TextChoices):
    """Tipos principales de cuentas"""
    ASSET = 'A', 'Activo'
    LIABILITY = 'P', 'Pasivo'
    EQUITY = 'E', 'Patrimonio'
    INCOME = 'I', 'Ingreso'
    EXPENSE = 'G', 'Gasto'

class AccountNature(models.TextChoices):
    """Naturaleza de la cuenta (determina su comportamiento)"""
    DEBIT = 'D', 'Deudora'
    CREDIT = 'C', 'Acreedora'

class Account(models.Model):
    """
    Modelo principal para el Plan de Cuentas.
    Implementa una estructura jerárquica recursiva.
    """
    code = models.CharField(
        max_length=20,
        unique=True,
        help_text="Código único de la cuenta (ej: 1.1.1.01)"
    )
    name = models.CharField(
        max_length=100,
        help_text="Nombre descriptivo de la cuenta"
    )
    level = models.CharField(
        max_length=1,
        choices=AccountLevel.choices,
        help_text="Nivel jerárquico de la cuenta"
    )
    type = models.CharField(
        max_length=1,
        choices=AccountType.choices,
        help_text="Tipo de cuenta"
    )
    nature = models.CharField(
        max_length=1,
        choices=AccountNature.choices,
        help_text="Naturaleza de la cuenta (Deudora/Acreedora)"
    )
    parent = models.ForeignKey(
        'self',
        null=True,
        blank=True,
        on_delete=models.PROTECT,
        related_name='children'
    )
    is_transactional = models.BooleanField(
        default=False,
        help_text="Indica si la cuenta puede recibir transacciones"
    )
    balance = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=Decimal('0.00'),
        validators=[
            MinValueValidator(Decimal('-999999999999.99')),
            MaxValueValidator(Decimal('999999999999.99'))
        ],
        help_text="Saldo actual de la cuenta"
    )
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    active = models.BooleanField(default=True)

    class Meta:
        verbose_name = 'Cuenta Contable'
        verbose_name_plural = 'Cuentas Contables'
        ordering = ['code']

    def __str__(self):
        return f"{self.code} - {self.name}"

    def clean(self):
        """Validaciones de negocio para la cuenta"""
        self._validate_hierarchy()
        self._validate_code_format()
        self._validate_account_type()
        self._validate_transaction_rules()

    def _validate_hierarchy(self):
        """Valida la jerarquía de la cuenta"""
        if self.parent:
            # Validar nivel jerárquico
            level_order = {
                'G': 0, 'C': 1, 'A': 2, 'S': 3
            }
            if level_order[self.level] <= level_order[self.parent.level]:
                raise ValidationError(
                    'El nivel jerárquico debe ser mayor al del padre'
                )

            # Validar tipo de cuenta
            if self.type != self.parent.type:
                raise ValidationError(
                    'El tipo de cuenta debe coincidir con el de su padre'
                )
        else:
            # Solo grupos pueden no tener padre
            if self.level != 'G':
                raise ValidationError(
                    'Solo los grupos pueden no tener cuenta padre'
                )

    def _validate_code_format(self):
        """Valida el formato del código según el nivel"""
        code_patterns = {
            'G': r'^\d$',                    # Un dígito: 1
            'C': r'^\d\.\d{2}$',            # Tres dígitos: 1.01
            'A': r'^\d\.\d{2}\.\d{2}$',     # Cinco dígitos: 1.01.01
            'S': r'^\d\.\d{2}\.\d{2}\.\d{2}$' # Siete dígitos: 1.01.01.01
        }

        if not re.match(code_patterns[self.level], self.code):
            raise ValidationError(
                f'Formato de código inválido para nivel {self.get_level_display()}'
            )

    def _validate_account_type(self):
        """Valida la naturaleza según el tipo de cuenta"""
        nature_by_type = {
            'A': 'D',  # Activos son deudores
            'P': 'C',  # Pasivos son acreedores
            'E': 'C',  # Patrimonio es acreedor
            'I': 'C',  # Ingresos son acreedores
            'G': 'D'   # Gastos son deudores
        }

        if self.nature != nature_by_type[self.type]:
            raise ValidationError(
                f'La naturaleza no corresponde al tipo de cuenta'
            )

    def _validate_transaction_rules(self):
        """Valida reglas para cuentas transaccionales"""
        if self.is_transactional:
            if self.level != 'S':
                raise ValidationError(
                    'Solo las subcuentas pueden ser transaccionales'
                )
            if self.children.exists():
                raise ValidationError(
                    'Una cuenta transaccional no puede tener subcuentas'
                )
        elif self.level == 'S':
            raise ValidationError(
                'Las subcuentas deben ser transaccionales'
            )

    def update_balance(self):
        """Actualiza el saldo de la cuenta y propaga hacia arriba"""
        if self.is_transactional:
            # Calcular saldo desde las transacciones
            transactions = self.transactions.all()
            debits = sum(t.amount for t in transactions if t.entry_type == 'D')
            credits = sum(t.amount for t in transactions if t.entry_type == 'C')

            # Aplicar según naturaleza de la cuenta
            if self.nature == 'D':
                self.balance = debits - credits
            else:
                self.balance = credits - debits
        else:
            # Para cuentas no transaccionales, sumar saldos de hijos
            self.balance = sum(child.balance for child in self.children.all())

        self.save()

        # Propagar hacia arriba
        if self.parent:
            self.parent.update_balance()

    @property
    def absolute_balance(self):
        """Retorna el saldo considerando la naturaleza de la cuenta"""
        if self.nature == 'D':
            return self.balance
        return -self.balance

    def get_descendants(self):
        """Retorna todas las cuentas descendientes"""
        descendants = []
        for child in self.children.all():
            descendants.append(child)
            descendants.extend(child.get_descendants())
        return descendants

    def get_ancestors(self):
        """Retorna todas las cuentas ancestras"""
        ancestors = []
        parent = self.parent
        while parent:
            ancestors.append(parent)
            parent = parent.parent
        return ancestors

Configuración del Admin

from django.contrib import admin
from .models import Account

@admin.register(Account)
class AccountAdmin(admin.ModelAdmin):
    list_display = [
        'code', 'name', 'level', 'type', 'nature',
        'is_transactional', 'balance', 'active'
    ]
    list_filter = ['level', 'type', 'nature', 'is_transactional', 'active']
    search_fields = ['code', 'name']
    readonly_fields = ['balance', 'created_at', 'updated_at']

    def get_readonly_fields(self, request, obj=None):
        if obj:  # editing an existing object
            return self.readonly_fields + ['code', 'level', 'type', 'nature']
        return self.readonly_fields

    def get_queryset(self, request):
        return super().get_queryset(request).select_related('parent')

Tests Unitarios

from django.test import TestCase
from django.core.exceptions import ValidationError
from decimal import Decimal
from .models import Account

class AccountTests(TestCase):
    def setUp(self):
        # Crear estructura básica
        self.assets = Account.objects.create(
            code='1',
            name='Activos',
            level='G',
            type='A',
            nature='D'
        )

        self.current_assets = Account.objects.create(
            code='1.01',
            name='Activos Corrientes',
            level='C',
            type='A',
            nature='D',
            parent=self.assets
        )

    def test_hierarchy_validation(self):
        """Prueba las validaciones jerárquicas"""
        with self.assertRaises(ValidationError):
            # Intentar crear una clase sin padre
            Account.objects.create(
                code='2.01',
                name='Invalid',
                level='C',
                type='A',
                nature='D'
            )

    def test_code_format(self):
        """Prueba la validación de formato de código"""
        with self.assertRaises(ValidationError):
            Account.objects.create(
                code='1.1',  # Formato inválido para clase
                name='Invalid',
                level='C',
                type='A',
                nature='D',
                parent=self.assets
            )

    def test_balance_propagation(self):
        """Prueba la propagación de saldos"""
        # Crear cuenta transaccional
        cash = Account.objects.create(
            code='1.01.01.01',
            name='Caja General',
            level='S',
            type='A',
            nature='D',
            parent=self.current_assets,
            is_transactional=True
        )

        # Simular transacción
        cash.balance = Decimal('1000.00')
        cash.save()
        cash.update_balance()

        # Verificar propagación
        self.current_assets.refresh_from_db()
        self.assets.refresh_from_db()

        self.assertEqual(self.current_assets.balance, Decimal('1000.00'))
        self.assertEqual(self.assets.balance, Decimal('1000.00'))

Ejemplo Real: Creación del Plan de Cuentas Base

def create_base_chart_of_accounts():
    """Crea una estructura básica del plan de cuentas"""
    # Activos
    assets = Account.objects.create(
        code='1',
        name='Activos',
        level='G',
        type='A',
        nature='D'
    )

    current_assets = Account.objects.create(
        code='1.01',
        name='Activos Corrientes',
        level='C',
        type='A',
        nature='D',
        parent=assets
    )

    cash = Account.objects.create(
        code='1.01.01',
        name='Efectivo y Equivalentes',
        level='A',
        type='A',
        nature='D',
        parent=current_assets
    )

    Account.objects.create(
        code='1.01.01.01',
        name='Caja General',
        level='S',
        type='A',
        nature='D',
        parent=cash,
        is_transactional=True
    )

    # Pasivos
    liabilities = Account.objects.create(
        code='2',
        name='Pasivos',
        level='G',
        type='P',
        nature='C'
    )

    # ... continuar con la estructura

Mejores Prácticas

Validaciones de Seguridad
- Proteger la integridad jerárquica
- Validar formatos de código
- Controlar la naturaleza de las cuentas
- Prevenir modificaciones no autorizadas
Manejo de Errores
- Validaciones específicas por nivel
- Mensajes de error claros
- Control de saldos y balances
Patrones de Diseño
- Composite para estructura jerárquica
- Observer para actualización de saldos
- Template Method para validaciones

Conclusión

Has aprendido a implementar un Plan de Cuentas que:

Mantiene una estructura jerárquica sólida
Valida la integridad de los datos contables
Propaga correctamente los saldos
Permite una gestión eficiente desde el Admin de Django
Implementa validaciones robustas de negocio
Mantiene la consistencia de los datos contables

Contabilidad para Django Developers: Implementando Partida Doble

Enrique Lazo Bello — Fri, 08 Nov 2024 16:38:35 +0000

Introducción

¿Has intentado implementar un sistema contable y te has encontrado con términos como "debe", "haber" y "partida doble" que parecen escritos en otro idioma? Como desarrollador, probablemente estés familiarizado con las transacciones ACID en bases de datos, donde todo debe mantenerse consistente. La partida doble en contabilidad funciona de manera similar: cada operación debe mantener el sistema en equilibrio.

En este tutorial, aprenderás a implementar un sistema de partida doble en Django, traduciendo conceptos contables a términos de programación que ya conoces. Al final, podrás crear un sistema contable robusto que garantice la integridad de los datos financieros.

Prerrequisitos

Python 3.12
Django 5.0
Conocimientos básicos de modelos en Django
Un editor de código (VS Code, PyCharm, etc.)

Conceptos Clave: Contabilidad para Developers

La Partida Doble Explicada con Git y Bases de Datos

Analogía #1: Git Commits

# En Git:
# Un commit siempre involucra dos operaciones equilibradas:
git add file.py    # (+100 bytes)
git rm old.py      # (-100 bytes)
# El repositorio mantiene su balance

# En Contabilidad:
# Un asiento siempre involucra dos o más cuentas que se equilibran:
Banco += 1000      # Débito
Capital -= 1000    # Crédito
# Las cuentas mantienen su balance

Analogía #2: Transacciones SQL

-- En una transacción bancaria:
BEGIN TRANSACTION;
    UPDATE cuenta_origen SET balance = balance - 100;
    UPDATE cuenta_destino SET balance = balance + 100;
COMMIT;

-- En contabilidad es similar:
BEGIN TRANSACTION;
    INSERT INTO asiento_contable (
        cuenta='Banco', tipo='DEBITO', monto=100);
    INSERT INTO asiento_contable (
        cuenta='Capital', tipo='CREDITO', monto=100);
COMMIT;

Implementación en Django

1. Modelos Base

from django.db import models
from django.core.exceptions import ValidationError
from django.db import transaction
from decimal import Decimal

class Account(models.Model):
    """
    Representa una cuenta contable (ej: Banco, Caja, Capital)
    """
    ACCOUNT_TYPES = [
        ('ASSET', 'Activo'),      # Aumenta con débito
        ('LIABILITY', 'Pasivo'),  # Aumenta con crédito
        ('EQUITY', 'Capital'),    # Aumenta con crédito
        ('INCOME', 'Ingreso'),    # Aumenta con crédito
        ('EXPENSE', 'Gasto')      # Aumenta con débito
    ]

    code = models.CharField(
        max_length=20, 
        unique=True,
        help_text="Código único de la cuenta"
    )
    name = models.CharField(max_length=100)
    type = models.CharField(
        max_length=10,
        choices=ACCOUNT_TYPES
    )
    balance = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=0,
        editable=False
    )

    @property
    def increases_with_debit(self):
        """Determina si la cuenta aumenta con débito"""
        return self.type in ['ASSET', 'EXPENSE']

    def update_balance(self, debit=0, credit=0):
        """Actualiza el balance según el tipo de cuenta"""
        net_change = debit - credit
        if not self.increases_with_debit:
            net_change = -net_change
        self.balance += Decimal(str(net_change))
        self.save()

    def __str__(self):
        return f"{self.code} - {self.name}"

    class Meta:
        verbose_name = "Cuenta"
        verbose_name_plural = "Cuentas"

class JournalEntry(models.Model):
    """
    Representa un asiento contable (conjunto de movimientos balanceados)
    """
    date = models.DateField()
    reference = models.CharField(max_length=50)
    description = models.TextField()
    is_posted = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)

    def clean(self):
        """Valida que el asiento esté balanceado"""
        if self.pk:  # Solo validar si ya existe
            total_debit = sum(
                line.debit for line in self.lines.all())
            total_credit = sum(
                line.credit for line in self.lines.all())

            if total_debit != total_credit:
                raise ValidationError(
                    'El total de débitos debe igual al total de créditos. '
                    f'Débitos: {total_debit}, Créditos: {total_credit}'
                )

    @transaction.atomic
    def post(self):
        """Aplica el asiento actualizando los balances"""
        if self.is_posted:
            raise ValidationError('Este asiento ya fue aplicado')

        self.clean()  # Validar balance

        for line in self.lines.all():
            line.account.update_balance(
                debit=line.debit,
                credit=line.credit
            )

        self.is_posted = True
        self.save()

    def __str__(self):
        return f"{self.date} - {self.reference}"

class JournalEntryLine(models.Model):
    """
    Representa una línea individual de un asiento contable
    """
    entry = models.ForeignKey(
        JournalEntry,
        related_name='lines',
        on_delete=models.CASCADE
    )
    account = models.ForeignKey(
        Account,
        on_delete=models.PROTECT
    )
    debit = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=0
    )
    credit = models.DecimalField(
        max_digits=15,
        decimal_places=2,
        default=0
    )
    description = models.CharField(max_length=200)

    def clean(self):
        """Valida que la línea tenga débito o crédito, pero no ambos"""
        if self.debit and self.credit:
            raise ValidationError(
                'Una línea no puede tener débito y crédito simultáneamente'
            )
        if not self.debit and not self.credit:
            raise ValidationError(
                'Debe especificar un valor de débito o crédito'
            )

    def __str__(self):
        return f"{self.account} - D:{self.debit} C:{self.credit}"

2. Configuración del Admin

from django.contrib import admin
from django.utils.html import format_html

@admin.register(Account)
class AccountAdmin(admin.ModelAdmin):
    list_display = ['code', 'name', 'type', 'formatted_balance']
    search_fields = ['code', 'name']
    readonly_fields = ['balance']

    def formatted_balance(self, obj):
        color = 'green' if obj.balance >= 0 else 'red'
        return format_html(
            '<span style="color: {};">${:,.2f}</span>',
            color,
            abs(obj.balance)
        )
    formatted_balance.short_description = 'Balance'

class JournalEntryLineInline(admin.TabularInline):
    model = JournalEntryLine
    extra = 2

@admin.register(JournalEntry)
class JournalEntryAdmin(admin.ModelAdmin):
    list_display = [
        'date', 
        'reference', 
        'description', 
        'is_posted'
    ]
    list_filter = ['date', 'is_posted']
    search_fields = ['reference', 'description']
    inlines = [JournalEntryLineInline]
    actions = ['post_entries']

    def post_entries(self, request, queryset):
        """Acción para aplicar múltiples asientos"""
        for entry in queryset.filter(is_posted=False):
            try:
                entry.post()
            except ValidationError as e:
                self.message_user(
                    request,
                    f'Error en asiento {entry}: {str(e)}',
                    level='ERROR'
                )
                return

        self.message_user(
            request,
            'Asientos aplicados correctamente'
        )
    post_entries.short_description = "Aplicar asientos seleccionados"

3. Tests

from django.test import TestCase
from django.core.exceptions import ValidationError
from decimal import Decimal
from .models import Account, JournalEntry, JournalEntryLine

class AccountingTestCase(TestCase):
    def setUp(self):
        # Crear cuentas de prueba
        self.cash = Account.objects.create(
            code='1001',
            name='Caja',
            type='ASSET'
        )
        self.capital = Account.objects.create(
            code='3001',
            name='Capital',
            type='EQUITY'
        )

    def test_balanced_entry(self):
        """Prueba que un asiento balanceado se aplique correctamente"""
        entry = JournalEntry.objects.create(
            date='2024-01-01',
            reference='TEST-001',
            description='Aporte de capital'
        )

        # Crear líneas del asiento
        JournalEntryLine.objects.create(
            entry=entry,
            account=self.cash,
            debit=1000
        )
        JournalEntryLine.objects.create(
            entry=entry,
            account=self.capital,
            credit=1000
        )

        # El asiento debería aplicarse sin errores
        entry.post()

        # Verificar balances
        self.cash.refresh_from_db()
        self.capital.refresh_from_db()
        self.assertEqual(self.cash.balance, Decimal('1000'))
        self.assertEqual(self.capital.balance, Decimal('1000'))

    def test_unbalanced_entry(self):
        """Prueba que un asiento desbalanceado genere error"""
        entry = JournalEntry.objects.create(
            date='2024-01-01',
            reference='TEST-002',
            description='Asiento desbalanceado'
        )

        JournalEntryLine.objects.create(
            entry=entry,
            account=self.cash,
            debit=1000
        )
        JournalEntryLine.objects.create(
            entry=entry,
            account=self.capital,
            credit=900  # Desbalanceado intencionalmente
        )

        # Debería lanzar error al intentar aplicar
        with self.assertRaises(ValidationError):
            entry.post()

📱 Ejemplo Práctico: Transferencia entre Cuentas

# 1. Crear las cuentas necesarias
cuenta_origen = Account.objects.create(
    code='1001',
    name='Cuenta Corriente',
    type='ASSET'
)
cuenta_destino = Account.objects.create(
    code='1002',
    name='Cuenta Ahorro',
    type='ASSET'
)

# 2. Crear el asiento de transferencia
with transaction.atomic():
    asiento = JournalEntry.objects.create(
        date='2024-01-01',
        reference='TRANS-001',
        description='Transferencia entre cuentas'
    )

    # Débito a cuenta destino
    JournalEntryLine.objects.create(
        entry=asiento,
        account=cuenta_destino,
        debit=500,
        description='Recepción de fondos'
    )

    # Crédito a cuenta origen
    JournalEntryLine.objects.create(
        entry=asiento,
        account=cuenta_origen,
        credit=500,
        description='Envío de fondos'
    )

    # Aplicar el asiento
    asiento.post()

🔒 Mejores Prácticas

Validaciones de Seguridad
- Usar transaction.atomic() para operaciones múltiples
- Implementar permisos granulares en el Admin
- Validar montos negativos y ceros
Manejo de Errores
- Capturar y loguear errores específicos
- Implementar rollback automático
- Validar estados inconsistentes
Patrones de Diseño
- Repository Pattern para consultas complejas
- Observer Pattern para auditoría
- Strategy Pattern para cálculos específicos

Conclusión

Has aprendido a implementar un sistema de partida doble robusto en Django que:

Asegura que cada movimiento afecte dos o más cuentas
Valida que débitos y créditos estén balanceados
Mantiene la integridad de los datos financieros
Es fácil de auditar y mantener

Siguientes Pasos

Implementar reportes financieros
Agregar conciliación bancaria
Desarrollar un sistema de auditoría
Añadir validaciones específicas por tipo de cuenta

Referencias

Basic CSS and JavaScript Animations for an Interactive Web Page

Enrique Lazo Bello — Sat, 02 Mar 2024 04:14:19 +0000

Animations and interactive effects can bring a website to life. This article examines the code behind a simple web page that uses CSS animations and JavaScript to create an engaging user experience.

My Website Under Construction
The code we're analyzing is part of my new personal website, currently under construction. My aim is to implement various interactive and innovative effects using modern web technologies.

Currently, the site consists of a single page with CSS animations and a JavaScript-created interactive switch. Soon, I'll be adding new sections and functionalities.

If you're interested in this topic, I invite you to visit the site and subscribe to stay updated on new content. I'm documenting the development process on my blog, so stay tuned!

👉 Visit my website under construction: https://enlabedev.com/

Let's continue analyzing some of the implemented code...

CSS File

The style.css file contains the styles that shape the page's design and animations. Let's examine some key points:

:root {
  --light: #FFFFFF;
  --dark: #000;
  --red: #CC0000;
  --red-light: #FF3333;
  --gray-dark: #333333;
  --gray-light: #CCCCCC;
  --gray-lighter: #EEEEEE;
  --blue:#0066CC;
  --blue-light: #33CCFF;
  --animation-time: 2s;
  --width: 160px;
  --height: 160px; 
}

CSS Animations
Next, we define CSS animations for later use. For example, this animation moves a border horizontally:

@keyframes border-top {
  0% {
    left: 0;
  }
  50% {
    width: 100%;
  }
  100% {
    left: auto;
    right: 0;
  }
}
@keyframes border-right {
  0% {
    top: 0;
  }
  50% {
    height: 100%;
  }
  100% {
    top: auto;
    bottom: 0;
  }
}
@keyframes border-bottom {
  0% {
    right: 0;
  }
  50% {
    width: 100%;
  }
  100% {
    left: 0;
    right: auto;
  }
}
@keyframes border-left {
  0% {
    bottom: 0;
  }
  50% {
    height: 100%;
  }
  100% {
    top: 0;
    bottom: auto;
  }
}

Classes and Media Queries
There are also CSS classes like .d-block, .text-normal that allow for reusable styles. And media queries for responsive design:

@media (min-width: 768px) {
  .title {
    font-size: 2.5rem;
  }
  .roles__item {
    display: inline-block;
  }
}

HTML File

The index.html file contains the webpage's structure and content. The HTML follows a standard structure: DOCTYPE, html, head, body, etc.

Using HTML5 Tags for SEO
The code uses some HTML5 semantic tags that can improve the website's SEO:

Headings
Heading tags <h1> are used to define content hierarchy and structure

<h1 class="text-normal title">
    <span class="roles__item animate__animated animate__fadeInUp">Engineer</span>
    <span class="roles__item animate__animated animate__fadeInUp"><strong>Experience</strong></span>
    <span class="roles__item animate__animated animate__fadeInUp">Creativity</span>
</h1>

The .text-normal and .title classes standardize the header size. By default, h1, h2, h3... tags come bold, we only needed to bold the word experience and used it as a device

SVG inline
For icons, inline SVGs are used, which can be manipulated with CSS and JS:

<svg class="power-off">
  <use .../>
</svg>

JavaScript File

Finally, the script.js file contains the JavaScript code that adds interactivity to the page
Event Listeners
Event listeners are added to the switch and body

powerSwitch.addEventListener('click', function() {
    powerSwitch.classList.toggle('power-off');
    powerSwitch.classList.toggle('power-on');
    body.classList.toggle('dark');
    body.classList.toggle('light');
  });

Timers
Timers (setTimeout) are used to sequence animations and transitions

setTimeout(() => {
  // code executed after X seconds 
}, 500)

Class Manipulation
CSS classes are added/removed using classList to toggle styles and animations.

element.classList.remove('animate__fadeInUp')
element.classList.add('animate__fadeOutDown')

Promises
Promises are used to chain effects asynchronously and avoid timing issues.

doAnim1()
  .then(() => doAnim2())
  .then(() => doAnim3())

In Summary

CSS Animations: We've defined key animations that enable visual effects such as moving and changing borders, creating an attractive dynamic around the central logo. These animations are crucial for capturing the user's attention from the outset.

Power On/Off Effect: Through JavaScript, we've implemented an interactive switch that toggles between light and dark modes. This effect is not only a central part of the user experience but also demonstrates modern web development practices to enhance site accessibility and visual comfort.

Responsive Design: By using media queries and reusable CSS classes, we ensure the site looks and functions perfectly across a wide range of devices. This underscores the importance of adaptive web design in today's digital world.

JavaScript Interactivity: The use of event listeners and CSS class manipulation through JavaScript allows us to create an interactive webpage. From sequential animations to color mode transitions, each element has been thoughtfully designed to enhance the user experience.

Accessibility and SEO: Through the use of HTML5 semantic tags and a clear content structure, we focus on accessibility and search engine optimization (SEO), ensuring our site is not only attractive but also functional and easy to find.

Conclusions

This article analyzed the source code behind an interactive web page with CSS animations and JavaScript.

We saw how elements like CSS variables, keyframe animations, transitions, reusable classes, and media queries bring design to life in an adaptable manner.

We also explored the use of HTML5 semantic tags, event listeners, timers, and promises in JavaScript to achieve engaging user experiences.

Performance, accessibility, and inclusion considerations are key when building modern, interactive websites.

I invite you to explore the complete source code at this GitHub repository:

https://github.com/enlabedev/personal-site

With these basic concepts, you can start experimenting and creating your own interactive effects on websites. Feel free to learn more and share your creations! You can find me at @enlabedev.

DEV Community: Enrique Lazo Bello

🚀 Notificaciones en Tiempo Real con SSE, FastAPI y Celery

¿Cuándo usar SSE y cuándo no?

La Arquitectura: Por qué 4 piezas y no 2

Paso 1: El Backend (FastAPI)

El endpoint de encolar

El stream SSE: Máxima estabilidad con StreamingResponse

Paso 2: El Worker (Celery)

Paso 3: El Frontend (React)

Paso 4: Dockerización

Casos de Producción: Lo que no te dicen

Conclusión

¿Cómo diseño APIs REST limpias y fáciles de mantener (Python/Laravel)?

Estructurando nuestro proyecto

Validar todo desde el lado del cliente

Versionado simple

Manejo explícito de errores: hablar el idioma correcto

Logging y trazabilidad: debugging sin adivinar

Evitar sobrecargar los endpoints: una responsabilidad por endpoint

Ejemplo práctico: estructura completa de un endpoint real

Conclusión

Contabilidad para Django Developers: Implementando Reportería Regulatoria SBS

Introducción

Prerrequisitos

Conceptos Clave para Developers

La Contabilidad como Sistema de Base de Datos

Implementación Práctica

1. Modelos Base

2. Admin personalizado

3. Tests Unitarios

Ejemplo Real: Sistema de Transferencias

Mejores Prácticas

Conclusión

Contabilidad para Django Developers: Implementando Libros Contables

Introducción

Prerrequisitos

Conceptos Clave

La Contabilidad desde una Perspectiva de Desarrollador

Implementación Práctica

Modelos Django

Configuración del Admin

Tests Unitarios

Ejemplo Real: Sistema de Transferencias entre Cuentas

Crear una Transferencia Bancaria

Mejores Prácticas

1. Validaciones de Seguridad

2. Manejo de Errores

3. Patrones de Diseño Recomendados

Command Pattern para Operaciones Contables

Queries Útiles

1. Balance por Tipo de Cuenta

2. Libro Mayor

Conclusión

Contabilidad para Django Developers: Implementando Reglas de Negocio Contables

Introducción

Prerrequisitos

Conceptos Clave

1. Asientos Contables = Transacciones Atómicas

2. Período Contable = Ventana de Tiempo Válida

3. Plan de Cuentas = Enum Jerárquico

Implementación Práctica

Ejemplo Real: Sistema de Transferencias entre Cuentas

Mejores Prácticas

1. Validaciones de Seguridad

2. Manejo de Errores

3. Patrones de Diseño Recomendados

Consultas Útiles

Conclusión

Contabilidad para Django Developers: Control de Cambios y Tipo de Cambio

Introducción

Prerrequisitos

Conceptos Clave

El Tipo de Cambio como un State Pattern

Diferencia de Cambio como Debugging

Implementación Práctica

Modelos Base

Configuración del Admin

Tests Unitarios

Ejemplo Real: Sistema de Transferencias

Caso de Uso: Transferencia entre Cuentas

El stream SSE: Máxima estabilidad con `StreamingResponse`