feat(fin): datos fiscales del emisor por empresa

fin.issuer_settings guarda la identidad fiscal con la que la empresa emite
CFDI: razón social, RFC, régimen fiscal y CP del lugar de expedición.

Una sola configuración vigente por empresa, garantizada con índice único
parcial; el guardado es un upsert (GET + PUT, sin DELETE). El RFC se valida con
la expresión oficial y se normaliza a mayúsculas sin espacios antes de aplicar
la restricción de longitud.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-08-07 16:57:39 -05:00
parent 9cf142add6
commit b8b8311ece
5 changed files with 209 additions and 0 deletions

View File

@@ -0,0 +1 @@
"""Datos fiscales del emisor por empresa."""

View File

@@ -0,0 +1,60 @@
"""Esquemas de los datos fiscales del emisor."""
import re
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field, field_validator
from ..catalogs.dto import TaxRegimeResponse
# RFC de persona moral (3 letras) o física (4 letras) + fecha + homoclave.
RFC_PATTERN = re.compile(r"^[A-ZÑ&]{3,4}\d{6}[A-Z0-9]{3}$")
ZIP_PATTERN = re.compile(r"^\d{5}$")
class IssuerSettingsInput(BaseModel):
"""Alta o actualización de los datos fiscales del emisor."""
legal_name: str = Field(..., min_length=1, max_length=255, description="Razón social")
rfc: str = Field(..., max_length=13, description="RFC del emisor")
tax_regime_id: int = Field(..., description="Régimen fiscal (c_RegimenFiscal)")
zip_code: str | None = Field(None, max_length=5, description="CP del lugar de expedición")
# mode="before": la normalización corre antes que el max_length del campo, para que
# un RFC con espacios de sobra no se rechace por longitud antes de limpiarlo.
@field_validator("rfc", mode="before")
@classmethod
def _validate_rfc(cls, value: str) -> str:
"""Normaliza a mayúsculas sin espacios y valida el formato oficial del RFC."""
if not isinstance(value, str):
raise ValueError("El RFC debe ser texto")
normalized = value.replace(" ", "").replace("-", "").upper()
if not RFC_PATTERN.match(normalized):
raise ValueError("El RFC no tiene un formato válido (ej. XAXX010101000)")
return normalized
@field_validator("zip_code")
@classmethod
def _validate_zip(cls, value: str | None) -> str | None:
if value is None or value == "":
return None
normalized = value.strip()
if not ZIP_PATTERN.match(normalized):
raise ValueError("El código postal debe tener 5 dígitos")
return normalized
class IssuerSettingsResponse(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
tenant_id: int
company_id: int
legal_name: str
rfc: str
tax_regime_id: int
tax_regime: TaxRegimeResponse | None = None
zip_code: str | None = None
updated_by: str | None = None
created_at: datetime
updated_at: datetime

View File

@@ -0,0 +1,42 @@
"""Datos fiscales del emisor — ``fin.issuer_settings``.
Es la identidad fiscal con la que la empresa emite CFDI: razón social, RFC, régimen
fiscal y código postal del lugar de expedición. Hay **una sola configuración vigente
por empresa**, garantizada con un índice único parcial.
"""
from sqlalchemy import ForeignKey, Index, Integer, String, 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 TaxRegime # noqa: F401 (resuelve la relación)
_ALIVE = text("deleted_at IS NULL")
class IssuerSettings(Base, TenantScopedMixin, TimestampMixin):
"""Configuración fiscal del emisor de la empresa."""
__tablename__ = "issuer_settings"
__table_args__ = (
Index(
"uq_fin_issuer_settings_company",
"tenant_id", "company_id",
unique=True, postgresql_where=_ALIVE, sqlite_where=_ALIVE,
),
{"schema": "fin"},
)
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
legal_name: Mapped[str] = mapped_column(String(255), nullable=False) # razón social
rfc: Mapped[str] = mapped_column(String(13), nullable=False)
tax_regime_id: Mapped[int] = mapped_column(
Integer, ForeignKey("sat.tax_regimes.id"), nullable=False, index=True
)
# CP del lugar de expedición del comprobante
zip_code: Mapped[str | None] = mapped_column(String(5), nullable=True)
updated_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
tax_regime: Mapped["TaxRegime"] = relationship("TaxRegime", lazy="selectin")

View File

@@ -0,0 +1,48 @@
"""Endpoints de los datos fiscales del emisor (una configuración por empresa)."""
from fastapi import APIRouter, Depends, Query
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 IssuerSettingsInput, IssuerSettingsResponse
router = APIRouter()
@router.get(
"/settings/issuer",
response_model=IssuerSettingsResponse,
dependencies=[Depends(PermissionChecker(["fin.settings.view"]))],
)
def get_issuer_settings(
company_id: int = Query(..., description="Company ID"),
current_user: dict = Depends(get_current_user),
db: Session = Depends(get_core_db),
):
"""Devuelve 404 mientras la empresa no haya capturado sus datos fiscales."""
return service.get_issuer_settings(db, current_user["tenant_id"], company_id)
@router.put(
"/settings/issuer",
response_model=IssuerSettingsResponse,
dependencies=[Depends(PermissionChecker(["fin.settings.edit"]))],
)
def save_issuer_settings(
payload: IssuerSettingsInput,
company_id: int = Query(..., description="Company ID"),
current_user: dict = Depends(get_current_user),
db: Session = Depends(get_core_db),
):
"""Alta o actualización (upsert) de los datos fiscales del emisor."""
return service.save_issuer_settings(
db,
payload,
current_user["tenant_id"],
company_id,
current_user.get("sub") or current_user.get("id"),
)

View File

@@ -0,0 +1,58 @@
"""Lógica de los datos fiscales del emisor.
Una empresa tiene, a lo más, una configuración vigente: el guardado es un upsert, no
un alta que pueda duplicar filas.
"""
from fastapi import HTTPException, status
from sqlalchemy.orm import Session
from ..catalogs.models import TaxRegime
from .dto import IssuerSettingsInput
from .models import IssuerSettings
def _find(db: Session, tenant_id: int, company_id: int) -> IssuerSettings | None:
return db.query(IssuerSettings).filter(
IssuerSettings.tenant_id == tenant_id,
IssuerSettings.company_id == company_id,
IssuerSettings.deleted_at.is_(None),
).first()
def get_issuer_settings(db: Session, tenant_id: int, company_id: int) -> IssuerSettings:
obj = _find(db, tenant_id, company_id)
if not obj:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="La empresa aún no tiene datos fiscales del emisor configurados",
)
return obj
def save_issuer_settings(
db: Session,
payload: IssuerSettingsInput,
tenant_id: int,
company_id: int,
user_id: str | None = None,
) -> IssuerSettings:
"""Crea la configuración la primera vez y la actualiza en adelante."""
if db.query(TaxRegime.id).filter(TaxRegime.id == payload.tax_regime_id).first() is None:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="El régimen fiscal indicado no existe en el catálogo del SAT",
)
obj = _find(db, tenant_id, company_id)
data = payload.model_dump()
if obj is None:
obj = IssuerSettings(**data, tenant_id=tenant_id, company_id=company_id, updated_by=user_id)
db.add(obj)
else:
for field, value in data.items():
setattr(obj, field, value)
obj.updated_by = user_id
db.commit()
db.refresh(obj)
return obj