From 9cf142add6669a167bc06b55c06e9bad47798f7e Mon Sep 17 00:00:00 2001 From: Jair Cedillo Date: Fri, 7 Aug 2026 16:57:39 -0500 Subject: [PATCH] =?UTF-8?q?feat(fin):=20CRUD=20de=20conceptos=20con=20rela?= =?UTF-8?q?ci=C3=B3n=201:1=20a=20clave=20ProdServ?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit fin.concepts es el catálogo de conceptos facturables de cada empresa, ligado a una clave de producto/servicio del SAT. La relación es 1:1 por empresa: si dos conceptos compartieran la misma clave, al timbrar no habría forma de saber qué descripción corresponde. La unicidad se garantiza por índice único parcial (WHERE deleted_at IS NULL) y se valida además en el service para devolver 409 con mensaje en español en vez de un IntegrityError crudo. La baja lógica libera la clave y el código. Las respuestas traen los objetos del catálogo ya resueltos (selectin) para que el frontend no dispare N+1. Co-Authored-By: Claude Opus 5 (1M context) --- .../api/v1/modules/fin/concepts/__init__.py | 1 + backend/api/v1/modules/fin/concepts/dto.py | 55 +++++++ backend/api/v1/modules/fin/concepts/models.py | 67 ++++++++ backend/api/v1/modules/fin/concepts/routes.py | 96 ++++++++++++ .../api/v1/modules/fin/concepts/service.py | 146 ++++++++++++++++++ 5 files changed, 365 insertions(+) create mode 100644 backend/api/v1/modules/fin/concepts/__init__.py create mode 100644 backend/api/v1/modules/fin/concepts/dto.py create mode 100644 backend/api/v1/modules/fin/concepts/models.py create mode 100644 backend/api/v1/modules/fin/concepts/routes.py create mode 100644 backend/api/v1/modules/fin/concepts/service.py diff --git a/backend/api/v1/modules/fin/concepts/__init__.py b/backend/api/v1/modules/fin/concepts/__init__.py new file mode 100644 index 0000000..6fd4a40 --- /dev/null +++ b/backend/api/v1/modules/fin/concepts/__init__.py @@ -0,0 +1 @@ +"""Catálogo de conceptos de facturación por empresa.""" diff --git a/backend/api/v1/modules/fin/concepts/dto.py b/backend/api/v1/modules/fin/concepts/dto.py new file mode 100644 index 0000000..49f9a62 --- /dev/null +++ b/backend/api/v1/modules/fin/concepts/dto.py @@ -0,0 +1,55 @@ +"""Esquemas del catálogo de conceptos de facturación.""" + +from datetime import datetime +from decimal import Decimal + +from pydantic import BaseModel, ConfigDict, Field + +from ..catalogs.dto import ProductServiceResponse, TaxObjectResponse, UnitOfMeasureResponse + + +class ConceptBase(BaseModel): + code: str = Field(..., min_length=1, max_length=40, description="Clave interna del concepto") + description: str = Field(..., min_length=1, max_length=500) + product_service_id: int = Field(..., description="Clave ProdServ del SAT (1:1 por empresa)") + unit_of_measure_id: int | None = None + tax_object_id: int | None = None + unit_price: Decimal | None = Field(None, ge=0, max_digits=14, decimal_places=2) + currency: str = Field("MXN", min_length=3, max_length=3) + is_active: bool = True + notes: str | None = None + + +class ConceptCreate(ConceptBase): + pass + + +class ConceptUpdate(BaseModel): + """Actualización parcial: solo se tocan los campos enviados.""" + + code: str | None = Field(None, min_length=1, max_length=40) + description: str | None = Field(None, min_length=1, max_length=500) + product_service_id: int | None = None + unit_of_measure_id: int | None = None + tax_object_id: int | None = None + unit_price: Decimal | None = Field(None, ge=0, max_digits=14, decimal_places=2) + currency: str | None = Field(None, min_length=3, max_length=3) + is_active: bool | None = None + notes: str | None = None + + +class ConceptResponse(ConceptBase): + """Incluye los objetos del catálogo del SAT ya resueltos, para evitar N+1 en la UI.""" + + model_config = ConfigDict(from_attributes=True) + + id: int + tenant_id: int + company_id: int + product_service: ProductServiceResponse | None = None + unit_of_measure: UnitOfMeasureResponse | None = None + tax_object: TaxObjectResponse | None = None + created_by: str | None = None + updated_by: str | None = None + created_at: datetime + updated_at: datetime diff --git a/backend/api/v1/modules/fin/concepts/models.py b/backend/api/v1/modules/fin/concepts/models.py new file mode 100644 index 0000000..96838b5 --- /dev/null +++ b/backend/api/v1/modules/fin/concepts/models.py @@ -0,0 +1,67 @@ +"""Catálogo de conceptos de facturación — ``fin.concepts``. + +A diferencia de los catálogos del SAT, este es **propio de cada empresa**: cada +concepto que la empresa factura (flete internacional, despacho, almacenaje…) se +registra una vez y queda amarrado a la clave de producto/servicio del SAT que le +corresponde. + +La relación con ``sat.products_services`` es **1:1 por empresa**: si dos conceptos +compartieran la misma clave ProdServ, al timbrar no habría forma de saber cuál +descripción corresponde a la clave, así que la unicidad se garantiza por índice y se +valida además en el service para devolver un 409 con mensaje entendible. +""" + +from sqlalchemy import Boolean, ForeignKey, Index, Integer, Numeric, String, Text, text +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from api.v1.common.base_models import TenantScopedMixin, TimestampMixin +from core.database import Base + +from ..catalogs.models import ProductService, TaxObject, UnitOfMeasure # noqa: F401 (resuelve las relaciones) + +# Los índices son parciales (``WHERE deleted_at IS NULL``): un concepto dado de baja +# lógica libera su clave y su código para uno nuevo. +_ALIVE = text("deleted_at IS NULL") + + +class Concept(Base, TenantScopedMixin, TimestampMixin): + """Concepto facturable de una empresa, ligado a una clave ProdServ del SAT.""" + + __tablename__ = "concepts" + __table_args__ = ( + Index( + "uq_fin_concepts_code", + "tenant_id", "company_id", "code", + unique=True, postgresql_where=_ALIVE, sqlite_where=_ALIVE, + ), + Index( + "uq_fin_concepts_product_service", + "tenant_id", "company_id", "product_service_id", + unique=True, postgresql_where=_ALIVE, sqlite_where=_ALIVE, + ), + {"schema": "fin"}, + ) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + code: Mapped[str] = mapped_column(String(40), nullable=False) # clave interna del concepto + description: Mapped[str] = mapped_column(String(500), nullable=False) + product_service_id: Mapped[int] = mapped_column( + Integer, ForeignKey("sat.products_services.id"), nullable=False, index=True + ) + unit_of_measure_id: Mapped[int | None] = mapped_column( + Integer, ForeignKey("sat.units_of_measure.id"), nullable=True + ) + tax_object_id: Mapped[int | None] = mapped_column( + Integer, ForeignKey("sat.tax_objects.id"), nullable=True + ) + unit_price: Mapped[float | None] = mapped_column(Numeric(14, 2), nullable=True) + currency: Mapped[str] = mapped_column(String(3), nullable=False, server_default=text("'MXN'")) + is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("true")) + notes: Mapped[str | None] = mapped_column(Text, nullable=True) + created_by: Mapped[str | None] = mapped_column(String(64), nullable=True) + updated_by: Mapped[str | None] = mapped_column(String(64), nullable=True) + + # Cargadas con selectinload para que el listado no dispare N+1 consultas. + product_service: Mapped["ProductService"] = relationship("ProductService", lazy="selectin") + unit_of_measure: Mapped["UnitOfMeasure | None"] = relationship("UnitOfMeasure", lazy="selectin") + tax_object: Mapped["TaxObject | None"] = relationship("TaxObject", lazy="selectin") diff --git a/backend/api/v1/modules/fin/concepts/routes.py b/backend/api/v1/modules/fin/concepts/routes.py new file mode 100644 index 0000000..cce3e45 --- /dev/null +++ b/backend/api/v1/modules/fin/concepts/routes.py @@ -0,0 +1,96 @@ +"""Endpoints del catálogo de conceptos de facturación (CRUD por empresa).""" + +from fastapi import APIRouter, Depends, Query, status +from sqlalchemy.orm import Session + +from api.v1.modules.core.permissions.dependencies import PermissionChecker +from core.database import get_core_db +from core.security import get_current_user + +from . import service +from .dto import ConceptCreate, ConceptResponse, ConceptUpdate + +router = APIRouter() + + +def _uid(current_user: dict) -> str | None: + return current_user.get("sub") or current_user.get("id") + + +@router.get( + "/concepts", + response_model=list[ConceptResponse], + dependencies=[Depends(PermissionChecker(["fin.concept.view"]))], +) +def list_concepts( + company_id: int = Query(..., description="Company ID"), + search: str | None = Query(None, description="Búsqueda por clave o descripción"), + active_only: bool | None = Query(None, description="Filtra por conceptos activos o inactivos"), + product_service_id: int | None = Query(None, description="Filtra por clave ProdServ del SAT"), + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + return service.get_concepts( + db, current_user["tenant_id"], company_id, search, active_only, product_service_id + ) + + +@router.get( + "/concepts/{concept_id}", + response_model=ConceptResponse, + dependencies=[Depends(PermissionChecker(["fin.concept.view"]))], +) +def get_concept( + concept_id: int, + company_id: int = Query(..., description="Company ID"), + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + return service.get_concept(db, concept_id, current_user["tenant_id"], company_id) + + +@router.post( + "/concepts", + response_model=ConceptResponse, + status_code=status.HTTP_201_CREATED, + dependencies=[Depends(PermissionChecker(["fin.concept.create"]))], +) +def create_concept( + payload: ConceptCreate, + company_id: int = Query(..., description="Company ID"), + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + return service.create_concept(db, payload, current_user["tenant_id"], company_id, _uid(current_user)) + + +@router.patch( + "/concepts/{concept_id}", + response_model=ConceptResponse, + dependencies=[Depends(PermissionChecker(["fin.concept.edit"]))], +) +def update_concept( + concept_id: int, + payload: ConceptUpdate, + company_id: int = Query(..., description="Company ID"), + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + return service.update_concept( + db, concept_id, payload, current_user["tenant_id"], company_id, _uid(current_user) + ) + + +@router.delete( + "/concepts/{concept_id}", + status_code=status.HTTP_204_NO_CONTENT, + dependencies=[Depends(PermissionChecker(["fin.concept.delete"]))], +) +def delete_concept( + concept_id: int, + company_id: int = Query(..., description="Company ID"), + current_user: dict = Depends(get_current_user), + db: Session = Depends(get_core_db), +): + """Baja lógica del concepto (``deleted_at``).""" + service.delete_concept(db, concept_id, current_user["tenant_id"], company_id) diff --git a/backend/api/v1/modules/fin/concepts/service.py b/backend/api/v1/modules/fin/concepts/service.py new file mode 100644 index 0000000..c3aeb09 --- /dev/null +++ b/backend/api/v1/modules/fin/concepts/service.py @@ -0,0 +1,146 @@ +"""Lógica del catálogo de conceptos de facturación. + +Todas las consultas filtran por ``tenant_id``, ``company_id`` y ``deleted_at IS NULL``: +el catálogo es privado de cada empresa dentro de cada tenant. +""" + +from datetime import datetime, timezone + +from fastapi import HTTPException, status +from sqlalchemy import or_ +from sqlalchemy.orm import Session + +from ..catalogs.models import ProductService, TaxObject, UnitOfMeasure +from .dto import ConceptCreate, ConceptUpdate +from .models import Concept + + +def _check_sat_refs(db: Session, data: dict) -> None: + """Verifica que las claves del SAT referidas existan antes de guardar.""" + for field, model, msg in [ + ("product_service_id", ProductService, "La clave de producto/servicio del SAT no existe"), + ("unit_of_measure_id", UnitOfMeasure, "La unidad de medida del SAT no existe"), + ("tax_object_id", TaxObject, "El objeto de impuesto del SAT no existe"), + ]: + value = data.get(field) + if field in data and value is not None: + if db.query(model.id).filter(model.id == value).first() is None: + raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=msg) + + +def _check_unique( + db: Session, + tenant_id: int, + company_id: int, + code: str | None, + product_service_id: int | None, + exclude_id: int | None = None, +) -> None: + """Aplica en el service las mismas reglas que los índices únicos parciales. + + Sin esto el conflicto llegaría al cliente como un IntegrityError crudo; aquí se + traduce a un 409 con mensaje en español. + """ + base = db.query(Concept).filter( + Concept.tenant_id == tenant_id, + Concept.company_id == company_id, + Concept.deleted_at.is_(None), + ) + if exclude_id is not None: + base = base.filter(Concept.id != exclude_id) + + if code is not None and base.filter(Concept.code == code).first() is not None: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"Ya existe un concepto con la clave '{code}' en esta empresa", + ) + # Regla 1:1 — una clave ProdServ no puede repetirse entre conceptos de la empresa. + if product_service_id is not None and base.filter( + Concept.product_service_id == product_service_id + ).first() is not None: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="La clave de producto/servicio del SAT ya está asignada a otro concepto de esta empresa", + ) + + +def get_concepts( + db: Session, + tenant_id: int, + company_id: int, + search: str | None = None, + active_only: bool | None = None, + product_service_id: int | None = None, +) -> list[Concept]: + q = db.query(Concept).filter( + Concept.tenant_id == tenant_id, + Concept.company_id == company_id, + Concept.deleted_at.is_(None), + ) + if active_only is not None: + q = q.filter(Concept.is_active.is_(active_only)) + if product_service_id is not None: + q = q.filter(Concept.product_service_id == product_service_id) + if search: + term = f"%{search.strip()}%" + q = q.filter(or_(Concept.code.ilike(term), Concept.description.ilike(term))) + return q.order_by(Concept.code.asc()).all() + + +def get_concept(db: Session, concept_id: int, tenant_id: int, company_id: int) -> Concept: + obj = db.query(Concept).filter( + Concept.id == concept_id, + Concept.tenant_id == tenant_id, + Concept.company_id == company_id, + Concept.deleted_at.is_(None), + ).first() + if not obj: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Concepto no encontrado") + return obj + + +def create_concept( + db: Session, payload: ConceptCreate, tenant_id: int, company_id: int, user_id: str | None = None +) -> Concept: + data = payload.model_dump() + _check_sat_refs(db, data) + _check_unique(db, tenant_id, company_id, data["code"], data["product_service_id"]) + obj = Concept(**data, tenant_id=tenant_id, company_id=company_id, created_by=user_id, updated_by=user_id) + db.add(obj) + db.commit() + db.refresh(obj) + return obj + + +def update_concept( + db: Session, + concept_id: int, + payload: ConceptUpdate, + tenant_id: int, + company_id: int, + user_id: str | None = None, +) -> Concept: + obj = get_concept(db, concept_id, tenant_id, company_id) + data = payload.model_dump(exclude_unset=True) + _check_sat_refs(db, data) + _check_unique( + db, + tenant_id, + company_id, + data.get("code"), + data.get("product_service_id"), + exclude_id=obj.id, + ) + for field, value in data.items(): + setattr(obj, field, value) + obj.updated_by = user_id + db.commit() + db.refresh(obj) + return obj + + +def delete_concept(db: Session, concept_id: int, tenant_id: int, company_id: int) -> None: + """Baja lógica: libera la clave ProdServ y el código para un concepto nuevo.""" + obj = get_concept(db, concept_id, tenant_id, company_id) + obj.deleted_at = datetime.now(timezone.utc) + db.commit()