feat(fin): amarre de facturas y partidas a catálogos SAT
fin.invoices gana tipo de comprobante, forma y método de pago y CP de
expedición; fin.invoice_items gana concepto de catálogo y las claves ProdServ,
unidad y objeto de impuesto. Todas nullable: las facturas ya emitidas no las
tienen y siguen funcionando igual (listado, detalle, PDF, envío).
La columna de texto libre invoice_items.concept se conserva obligatoria porque
la consume el PDF actual; al capturar por catálogo, el service hereda ahí la
descripción del concepto cuando el cliente no la envía.
Nueva tabla fin.invoice_item_taxes para el detalle de impuestos trasladados y
retenidos por partida. No interviene en el cálculo de subtotal/IVA/total, que
sigue saliendo de invoices.tax_rate.
Incluye la migración e6f7a8b9c0d1 (crea el schema sat, siembra los catálogos con
sync_catalogs y monta las tablas e índices nuevos) y registra los permisos
fin.concept.* y fin.settings.{view,edit}.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,269 @@
|
||||
"""Catálogos SAT (schema sat), conceptos de facturación, datos fiscales del emisor
|
||||
y amarre de facturas y partidas a los catálogos.
|
||||
|
||||
Revision ID: e6f7a8b9c0d1
|
||||
Revises: d5e6f7a8b9c0
|
||||
Create Date: 2026-08-07 00:00:00.000000
|
||||
|
||||
"""
|
||||
from typing import Sequence, Union
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
from api.v1.modules.fin.catalogs.seed_data import sync_catalogs
|
||||
|
||||
revision: str = "e6f7a8b9c0d1"
|
||||
down_revision: Union[str, None] = "d5e6f7a8b9c0"
|
||||
branch_labels: Union[str, Sequence[str], None] = None
|
||||
depends_on: Union[str, Sequence[str], None] = None
|
||||
|
||||
# Índices únicos parciales: la baja lógica (deleted_at) libera la clave.
|
||||
_ALIVE = "deleted_at IS NULL"
|
||||
|
||||
# Catálogos del SAT: (tabla, longitud de code, columnas propias del catálogo).
|
||||
_SAT_CATALOGS: list[tuple[str, int, list[sa.Column]]] = [
|
||||
("tax_regimes", 3, [
|
||||
sa.Column("applies_to_individual", sa.Boolean(), nullable=False, server_default=sa.text("false")),
|
||||
sa.Column("applies_to_legal_entity", sa.Boolean(), nullable=False, server_default=sa.text("false")),
|
||||
]),
|
||||
("taxes", 3, [
|
||||
sa.Column("is_withholding", sa.Boolean(), nullable=False, server_default=sa.text("false")),
|
||||
sa.Column("is_transferred", sa.Boolean(), nullable=False, server_default=sa.text("false")),
|
||||
sa.Column("is_local", sa.Boolean(), nullable=False, server_default=sa.text("false")),
|
||||
]),
|
||||
("payment_forms", 2, []),
|
||||
("units_of_measure", 20, [
|
||||
sa.Column("name", sa.String(length=255), nullable=False),
|
||||
sa.Column("symbol", sa.String(length=20), nullable=True),
|
||||
]),
|
||||
("products_services", 8, []),
|
||||
("voucher_types", 1, []),
|
||||
("payment_methods", 3, []),
|
||||
("tax_objects", 2, []),
|
||||
]
|
||||
|
||||
# units_of_measure guarda el nombre corto aparte, así que su description es opcional.
|
||||
_NULLABLE_DESCRIPTION = {"units_of_measure"}
|
||||
|
||||
|
||||
def _timestamp_columns(with_soft_delete: bool) -> list[sa.Column]:
|
||||
columns = [
|
||||
sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")),
|
||||
sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")),
|
||||
]
|
||||
if with_soft_delete:
|
||||
columns.append(sa.Column("deleted_at", sa.DateTime(), nullable=True))
|
||||
return columns
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
# ---------- Schema y catálogos globales del SAT ----------
|
||||
op.execute("CREATE SCHEMA IF NOT EXISTS sat")
|
||||
|
||||
for table, code_length, extra_columns in _SAT_CATALOGS:
|
||||
op.create_table(
|
||||
table,
|
||||
sa.Column("id", sa.Integer(), nullable=False),
|
||||
sa.Column("code", sa.String(length=code_length), nullable=False),
|
||||
sa.Column(
|
||||
"description",
|
||||
sa.String(length=500),
|
||||
nullable=table in _NULLABLE_DESCRIPTION,
|
||||
),
|
||||
sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.text("true")),
|
||||
*extra_columns,
|
||||
*_timestamp_columns(with_soft_delete=False),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
schema="sat",
|
||||
)
|
||||
op.create_index(f"ix_sat_{table}_id", table, ["id"], schema="sat")
|
||||
# La clave oficial del SAT es única dentro de su catálogo.
|
||||
op.create_index(f"ix_sat_{table}_code", table, ["code"], unique=True, schema="sat")
|
||||
|
||||
# Semillas de los catálogos (idempotente: puede volver a correrse sin duplicar).
|
||||
sync_catalogs(op.get_bind())
|
||||
|
||||
# ---------- fin.concepts ----------
|
||||
op.create_table(
|
||||
"concepts",
|
||||
sa.Column("id", sa.Integer(), nullable=False),
|
||||
sa.Column("tenant_id", sa.Integer(), nullable=False),
|
||||
sa.Column("company_id", sa.Integer(), nullable=False),
|
||||
sa.Column("code", sa.String(length=40), nullable=False),
|
||||
sa.Column("description", sa.String(length=500), nullable=False),
|
||||
sa.Column("product_service_id", sa.Integer(), nullable=False),
|
||||
sa.Column("unit_of_measure_id", sa.Integer(), nullable=True),
|
||||
sa.Column("tax_object_id", sa.Integer(), nullable=True),
|
||||
sa.Column("unit_price", sa.Numeric(precision=14, scale=2), nullable=True),
|
||||
sa.Column("currency", sa.String(length=3), nullable=False, server_default=sa.text("'MXN'")),
|
||||
sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.text("true")),
|
||||
sa.Column("notes", sa.Text(), nullable=True),
|
||||
sa.Column("created_by", sa.String(length=64), nullable=True),
|
||||
sa.Column("updated_by", sa.String(length=64), nullable=True),
|
||||
*_timestamp_columns(with_soft_delete=True),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
sa.ForeignKeyConstraint(["tenant_id"], ["core.tenants.id"], name="fk_fin_concepts_tenant_id"),
|
||||
sa.ForeignKeyConstraint(
|
||||
["product_service_id"], ["sat.products_services.id"], name="fk_fin_concepts_product_service_id"
|
||||
),
|
||||
sa.ForeignKeyConstraint(
|
||||
["unit_of_measure_id"], ["sat.units_of_measure.id"], name="fk_fin_concepts_unit_of_measure_id"
|
||||
),
|
||||
sa.ForeignKeyConstraint(
|
||||
["tax_object_id"], ["sat.tax_objects.id"], name="fk_fin_concepts_tax_object_id"
|
||||
),
|
||||
schema="fin",
|
||||
)
|
||||
op.create_index("ix_fin_concepts_id", "concepts", ["id"], schema="fin")
|
||||
op.create_index("ix_fin_concepts_tenant_id", "concepts", ["tenant_id"], schema="fin")
|
||||
op.create_index("ix_fin_concepts_company_id", "concepts", ["company_id"], schema="fin")
|
||||
op.create_index("ix_fin_concepts_product_service_id", "concepts", ["product_service_id"], schema="fin")
|
||||
# La clave interna del concepto es única por empresa.
|
||||
op.create_index(
|
||||
"uq_fin_concepts_code", "concepts", ["tenant_id", "company_id", "code"],
|
||||
unique=True, schema="fin", postgresql_where=sa.text(_ALIVE),
|
||||
)
|
||||
# Relación 1:1 con c_ClaveProdServ: una clave del SAT no puede repetirse entre
|
||||
# los conceptos vigentes de la misma empresa.
|
||||
op.create_index(
|
||||
"uq_fin_concepts_product_service", "concepts", ["tenant_id", "company_id", "product_service_id"],
|
||||
unique=True, schema="fin", postgresql_where=sa.text(_ALIVE),
|
||||
)
|
||||
|
||||
# ---------- fin.issuer_settings ----------
|
||||
op.create_table(
|
||||
"issuer_settings",
|
||||
sa.Column("id", sa.Integer(), nullable=False),
|
||||
sa.Column("tenant_id", sa.Integer(), nullable=False),
|
||||
sa.Column("company_id", sa.Integer(), nullable=False),
|
||||
sa.Column("legal_name", sa.String(length=255), nullable=False),
|
||||
sa.Column("rfc", sa.String(length=13), nullable=False),
|
||||
sa.Column("tax_regime_id", sa.Integer(), nullable=False),
|
||||
sa.Column("zip_code", sa.String(length=5), nullable=True),
|
||||
sa.Column("updated_by", sa.String(length=64), nullable=True),
|
||||
*_timestamp_columns(with_soft_delete=True),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
sa.ForeignKeyConstraint(["tenant_id"], ["core.tenants.id"], name="fk_fin_issuer_settings_tenant_id"),
|
||||
sa.ForeignKeyConstraint(
|
||||
["tax_regime_id"], ["sat.tax_regimes.id"], name="fk_fin_issuer_settings_tax_regime_id"
|
||||
),
|
||||
schema="fin",
|
||||
)
|
||||
op.create_index("ix_fin_issuer_settings_id", "issuer_settings", ["id"], schema="fin")
|
||||
op.create_index("ix_fin_issuer_settings_tenant_id", "issuer_settings", ["tenant_id"], schema="fin")
|
||||
op.create_index("ix_fin_issuer_settings_company_id", "issuer_settings", ["company_id"], schema="fin")
|
||||
op.create_index("ix_fin_issuer_settings_tax_regime_id", "issuer_settings", ["tax_regime_id"], schema="fin")
|
||||
# Una sola configuración fiscal vigente por empresa.
|
||||
op.create_index(
|
||||
"uq_fin_issuer_settings_company", "issuer_settings", ["tenant_id", "company_id"],
|
||||
unique=True, schema="fin", postgresql_where=sa.text(_ALIVE),
|
||||
)
|
||||
|
||||
# ---------- fin.invoice_item_taxes ----------
|
||||
op.create_table(
|
||||
"invoice_item_taxes",
|
||||
sa.Column("id", sa.Integer(), nullable=False),
|
||||
sa.Column("tenant_id", sa.Integer(), nullable=False),
|
||||
sa.Column("company_id", sa.Integer(), nullable=False),
|
||||
sa.Column("invoice_item_id", sa.Integer(), nullable=False),
|
||||
sa.Column("tax_id", sa.Integer(), nullable=False),
|
||||
sa.Column("is_withholding", sa.Boolean(), nullable=False, server_default=sa.text("false")),
|
||||
sa.Column("rate", sa.Numeric(precision=8, scale=6), nullable=True),
|
||||
sa.Column("amount", sa.Numeric(precision=14, scale=2), nullable=False, server_default=sa.text("0")),
|
||||
*_timestamp_columns(with_soft_delete=True),
|
||||
sa.PrimaryKeyConstraint("id"),
|
||||
sa.ForeignKeyConstraint(["tenant_id"], ["core.tenants.id"], name="fk_fin_invoice_item_taxes_tenant_id"),
|
||||
sa.ForeignKeyConstraint(
|
||||
["invoice_item_id"], ["fin.invoice_items.id"], name="fk_fin_invoice_item_taxes_invoice_item_id"
|
||||
),
|
||||
sa.ForeignKeyConstraint(["tax_id"], ["sat.taxes.id"], name="fk_fin_invoice_item_taxes_tax_id"),
|
||||
schema="fin",
|
||||
)
|
||||
op.create_index("ix_fin_invoice_item_taxes_id", "invoice_item_taxes", ["id"], schema="fin")
|
||||
op.create_index("ix_fin_invoice_item_taxes_tenant_id", "invoice_item_taxes", ["tenant_id"], schema="fin")
|
||||
op.create_index("ix_fin_invoice_item_taxes_company_id", "invoice_item_taxes", ["company_id"], schema="fin")
|
||||
op.create_index(
|
||||
"ix_fin_invoice_item_taxes_invoice_item_id", "invoice_item_taxes", ["invoice_item_id"], schema="fin"
|
||||
)
|
||||
# Un mismo impuesto no puede declararse dos veces con el mismo rol en la partida.
|
||||
op.create_index(
|
||||
"uq_fin_invoice_item_taxes", "invoice_item_taxes", ["invoice_item_id", "tax_id", "is_withholding"],
|
||||
unique=True, schema="fin", postgresql_where=sa.text(_ALIVE),
|
||||
)
|
||||
|
||||
# ---------- fin.invoices: claves fiscales del comprobante ----------
|
||||
# Todas nullable: las facturas ya emitidas no tienen estos datos.
|
||||
op.add_column("invoices", sa.Column("voucher_type_id", sa.Integer(), nullable=True), schema="fin")
|
||||
op.add_column("invoices", sa.Column("payment_form_id", sa.Integer(), nullable=True), schema="fin")
|
||||
op.add_column("invoices", sa.Column("payment_method_id", sa.Integer(), nullable=True), schema="fin")
|
||||
op.add_column("invoices", sa.Column("expedition_zip_code", sa.String(length=5), nullable=True), schema="fin")
|
||||
op.create_foreign_key(
|
||||
"fk_fin_invoices_voucher_type_id", "invoices", "voucher_types",
|
||||
["voucher_type_id"], ["id"], source_schema="fin", referent_schema="sat",
|
||||
)
|
||||
op.create_foreign_key(
|
||||
"fk_fin_invoices_payment_form_id", "invoices", "payment_forms",
|
||||
["payment_form_id"], ["id"], source_schema="fin", referent_schema="sat",
|
||||
)
|
||||
op.create_foreign_key(
|
||||
"fk_fin_invoices_payment_method_id", "invoices", "payment_methods",
|
||||
["payment_method_id"], ["id"], source_schema="fin", referent_schema="sat",
|
||||
)
|
||||
|
||||
# ---------- fin.invoice_items: claves fiscales de la partida ----------
|
||||
# La columna de texto libre `concept` se conserva intacta y obligatoria: la usa el
|
||||
# PDF actual de la factura.
|
||||
op.add_column("invoice_items", sa.Column("concept_id", sa.Integer(), nullable=True), schema="fin")
|
||||
op.add_column("invoice_items", sa.Column("product_service_id", sa.Integer(), nullable=True), schema="fin")
|
||||
op.add_column("invoice_items", sa.Column("unit_of_measure_id", sa.Integer(), nullable=True), schema="fin")
|
||||
op.add_column("invoice_items", sa.Column("tax_object_id", sa.Integer(), nullable=True), schema="fin")
|
||||
op.create_index("ix_fin_invoice_items_concept_id", "invoice_items", ["concept_id"], schema="fin")
|
||||
op.create_foreign_key(
|
||||
"fk_fin_invoice_items_concept_id", "invoice_items", "concepts",
|
||||
["concept_id"], ["id"], source_schema="fin", referent_schema="fin",
|
||||
)
|
||||
op.create_foreign_key(
|
||||
"fk_fin_invoice_items_product_service_id", "invoice_items", "products_services",
|
||||
["product_service_id"], ["id"], source_schema="fin", referent_schema="sat",
|
||||
)
|
||||
op.create_foreign_key(
|
||||
"fk_fin_invoice_items_unit_of_measure_id", "invoice_items", "units_of_measure",
|
||||
["unit_of_measure_id"], ["id"], source_schema="fin", referent_schema="sat",
|
||||
)
|
||||
op.create_foreign_key(
|
||||
"fk_fin_invoice_items_tax_object_id", "invoice_items", "tax_objects",
|
||||
["tax_object_id"], ["id"], source_schema="fin", referent_schema="sat",
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
# fin.invoice_items
|
||||
for constraint in (
|
||||
"fk_fin_invoice_items_tax_object_id",
|
||||
"fk_fin_invoice_items_unit_of_measure_id",
|
||||
"fk_fin_invoice_items_product_service_id",
|
||||
"fk_fin_invoice_items_concept_id",
|
||||
):
|
||||
op.drop_constraint(constraint, "invoice_items", schema="fin", type_="foreignkey")
|
||||
op.drop_index("ix_fin_invoice_items_concept_id", table_name="invoice_items", schema="fin")
|
||||
for column in ("tax_object_id", "unit_of_measure_id", "product_service_id", "concept_id"):
|
||||
op.drop_column("invoice_items", column, schema="fin")
|
||||
|
||||
# fin.invoices
|
||||
for constraint in (
|
||||
"fk_fin_invoices_payment_method_id",
|
||||
"fk_fin_invoices_payment_form_id",
|
||||
"fk_fin_invoices_voucher_type_id",
|
||||
):
|
||||
op.drop_constraint(constraint, "invoices", schema="fin", type_="foreignkey")
|
||||
for column in ("expedition_zip_code", "payment_method_id", "payment_form_id", "voucher_type_id"):
|
||||
op.drop_column("invoices", column, schema="fin")
|
||||
|
||||
# Tablas nuevas (los índices caen con la tabla).
|
||||
op.drop_table("invoice_item_taxes", schema="fin")
|
||||
op.drop_table("issuer_settings", schema="fin")
|
||||
op.drop_table("concepts", schema="fin")
|
||||
|
||||
# Catálogos del SAT: se va el schema completo.
|
||||
op.execute("DROP SCHEMA IF EXISTS sat CASCADE")
|
||||
@@ -10,7 +10,16 @@ class InvoiceClientReviewInput(BaseModel):
|
||||
notes: str | None = None
|
||||
|
||||
|
||||
class InvoiceItemBase(BaseModel):
|
||||
class InvoiceItemSatFields(BaseModel):
|
||||
"""Claves fiscales de la partida. Opcionales: las facturas previas no las tienen."""
|
||||
|
||||
concept_id: int | None = None
|
||||
product_service_id: int | None = None
|
||||
unit_of_measure_id: int | None = None
|
||||
tax_object_id: int | None = None
|
||||
|
||||
|
||||
class InvoiceItemBase(InvoiceItemSatFields):
|
||||
concept: str = Field(..., max_length=60)
|
||||
description: str | None = Field(None, max_length=255)
|
||||
quantity: Decimal = Field(Decimal(1), ge=0, max_digits=12, decimal_places=2)
|
||||
@@ -19,9 +28,11 @@ class InvoiceItemBase(BaseModel):
|
||||
|
||||
class InvoiceItemCreate(InvoiceItemBase):
|
||||
invoice_id: int
|
||||
# Opcional solo si viene concept_id: el service copia la descripción del concepto.
|
||||
concept: str | None = Field(None, max_length=60)
|
||||
|
||||
|
||||
class InvoiceItemUpdate(BaseModel):
|
||||
class InvoiceItemUpdate(InvoiceItemSatFields):
|
||||
concept: str | None = Field(None, max_length=60)
|
||||
description: str | None = Field(None, max_length=255)
|
||||
quantity: Decimal | None = Field(None, ge=0, max_digits=12, decimal_places=2)
|
||||
@@ -76,6 +87,11 @@ class InvoiceBase(BaseModel):
|
||||
bank_info: str | None = None
|
||||
notes: str | None = None
|
||||
owner_user_id: str | None = Field(None, max_length=64)
|
||||
# ----- Claves fiscales del CFDI (opcionales mientras no se timbre) -----
|
||||
voucher_type_id: int | None = None
|
||||
payment_form_id: int | None = None
|
||||
payment_method_id: int | None = None
|
||||
expedition_zip_code: str | None = Field(None, max_length=5)
|
||||
|
||||
|
||||
class InvoiceCreate(InvoiceBase):
|
||||
@@ -94,6 +110,10 @@ class InvoiceUpdate(BaseModel):
|
||||
bank_info: str | None = None
|
||||
notes: str | None = None
|
||||
owner_user_id: str | None = Field(None, max_length=64)
|
||||
voucher_type_id: int | None = None
|
||||
payment_form_id: int | None = None
|
||||
payment_method_id: int | None = None
|
||||
expedition_zip_code: str | None = Field(None, max_length=5)
|
||||
|
||||
|
||||
class InvoiceResponse(InvoiceBase):
|
||||
|
||||
@@ -1,11 +1,22 @@
|
||||
from datetime import date, datetime
|
||||
|
||||
from sqlalchemy import Boolean, Date, DateTime, ForeignKey, Integer, Numeric, String, Text, text
|
||||
from sqlalchemy import Boolean, Date, DateTime, ForeignKey, Index, Integer, Numeric, String, Text, text
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
|
||||
from core.database import Base
|
||||
|
||||
from ..catalogs.models import ( # noqa: F401 (registra los catálogos SAT referidos por las FK)
|
||||
PaymentForm,
|
||||
PaymentMethod,
|
||||
ProductService,
|
||||
Tax,
|
||||
TaxObject,
|
||||
UnitOfMeasure,
|
||||
VoucherType,
|
||||
)
|
||||
from ..concepts.models import Concept # noqa: F401
|
||||
|
||||
|
||||
class Invoice(Base, TenantScopedMixin, TimestampMixin):
|
||||
"""Factura (Diagrama 4). Integra los costos de la operación para cobro al cliente."""
|
||||
@@ -50,6 +61,18 @@ class Invoice(Base, TenantScopedMixin, TimestampMixin):
|
||||
owner_user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
|
||||
created_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
updated_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
# ----- Datos fiscales del CFDI (catálogos SAT) -----
|
||||
# Nullables: las facturas emitidas antes de existir los catálogos no los tienen.
|
||||
voucher_type_id: Mapped[int | None] = mapped_column(
|
||||
Integer, ForeignKey("sat.voucher_types.id"), nullable=True
|
||||
)
|
||||
payment_form_id: Mapped[int | None] = mapped_column(
|
||||
Integer, ForeignKey("sat.payment_forms.id"), nullable=True
|
||||
)
|
||||
payment_method_id: Mapped[int | None] = mapped_column(
|
||||
Integer, ForeignKey("sat.payment_methods.id"), nullable=True
|
||||
)
|
||||
expedition_zip_code: Mapped[str | None] = mapped_column(String(5), nullable=True)
|
||||
|
||||
|
||||
class InvoiceItem(Base, TenantScopedMixin, TimestampMixin):
|
||||
@@ -62,10 +85,54 @@ class InvoiceItem(Base, TenantScopedMixin, TimestampMixin):
|
||||
invoice_id: Mapped[int] = mapped_column(
|
||||
Integer, ForeignKey("fin.invoices.id"), nullable=False, index=True
|
||||
)
|
||||
# Texto libre histórico: lo consume el PDF actual y se conserva obligatorio.
|
||||
concept: Mapped[str] = mapped_column(String(60), nullable=False)
|
||||
description: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
||||
quantity: Mapped[float] = mapped_column(Numeric(12, 2), nullable=False, server_default=text("1"))
|
||||
unit_amount: Mapped[float] = mapped_column(Numeric(14, 2), nullable=False, server_default=text("0"))
|
||||
# ----- Datos fiscales de la partida (catálogos SAT) -----
|
||||
concept_id: Mapped[int | None] = mapped_column(
|
||||
Integer, ForeignKey("fin.concepts.id"), nullable=True, index=True
|
||||
)
|
||||
product_service_id: Mapped[int | None] = mapped_column(
|
||||
Integer, ForeignKey("sat.products_services.id"), nullable=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
|
||||
)
|
||||
|
||||
|
||||
class InvoiceItemTax(Base, TenantScopedMixin, TimestampMixin):
|
||||
"""Impuesto trasladado o retenido de una partida de la factura.
|
||||
|
||||
Es captura de detalle fiscal para el futuro CFDI: **no** interviene en el cálculo
|
||||
de subtotal/IVA/total de la factura, que sigue saliendo de ``invoices.tax_rate``.
|
||||
"""
|
||||
|
||||
__tablename__ = "invoice_item_taxes"
|
||||
__table_args__ = (
|
||||
Index(
|
||||
"uq_fin_invoice_item_taxes",
|
||||
"invoice_item_id", "tax_id", "is_withholding",
|
||||
unique=True,
|
||||
postgresql_where=text("deleted_at IS NULL"),
|
||||
sqlite_where=text("deleted_at IS NULL"),
|
||||
),
|
||||
{"schema": "fin"},
|
||||
)
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
|
||||
invoice_item_id: Mapped[int] = mapped_column(
|
||||
Integer, ForeignKey("fin.invoice_items.id"), nullable=False, index=True
|
||||
)
|
||||
tax_id: Mapped[int] = mapped_column(Integer, ForeignKey("sat.taxes.id"), nullable=False)
|
||||
# false = trasladado (se cobra al cliente); true = retenido
|
||||
is_withholding: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("false"))
|
||||
rate: Mapped[float | None] = mapped_column(Numeric(8, 6), nullable=True) # p. ej. 0.160000
|
||||
amount: Mapped[float] = mapped_column(Numeric(14, 2), nullable=False, server_default=text("0"))
|
||||
|
||||
|
||||
class Payment(Base, TenantScopedMixin, TimestampMixin):
|
||||
|
||||
@@ -9,6 +9,7 @@ from api.v1.modules.crm.accounts.models import Account
|
||||
from api.v1.modules.crm.quotes.models import Quote, QuoteItem
|
||||
from api.v1.modules.ops.shipments.models import Shipment
|
||||
|
||||
from ..concepts.models import Concept
|
||||
from .dto import (
|
||||
InvoiceClientReviewInput,
|
||||
InvoiceCreate,
|
||||
@@ -331,9 +332,38 @@ def _get_item(db, item_id, tenant_id, company_id) -> InvoiceItem:
|
||||
return obj
|
||||
|
||||
|
||||
def _resolve_item_concept(db, data: dict, tenant_id, company_id) -> None:
|
||||
"""Completa ``concept`` a partir del concepto del catálogo cuando no se envió.
|
||||
|
||||
El PDF de la factura sigue leyendo la columna de texto libre ``concept``, así que
|
||||
al capturar por catálogo se hereda ahí la descripción del concepto (recortada al
|
||||
largo de la columna).
|
||||
"""
|
||||
concept_id = data.get("concept_id")
|
||||
if concept_id is not None:
|
||||
catalog_concept = 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 catalog_concept:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail="El concepto del catálogo no existe en esta empresa",
|
||||
)
|
||||
if not data.get("concept"):
|
||||
data["concept"] = catalog_concept.description[:60]
|
||||
if not data.get("concept"):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail="La partida requiere un concepto o una referencia al catálogo de conceptos",
|
||||
)
|
||||
|
||||
|
||||
def create_item(db, payload: InvoiceItemCreate, tenant_id, company_id) -> InvoiceItem:
|
||||
invoice = get_invoice(db, payload.invoice_id, tenant_id, company_id)
|
||||
item = InvoiceItem(**payload.model_dump(), tenant_id=tenant_id, company_id=company_id)
|
||||
data = payload.model_dump()
|
||||
_resolve_item_concept(db, data, tenant_id, company_id)
|
||||
item = InvoiceItem(**data, tenant_id=tenant_id, company_id=company_id)
|
||||
db.add(item)
|
||||
db.flush()
|
||||
_recompute(db, invoice)
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
from api.v1.modules.core.permissions.registry import registry
|
||||
|
||||
MODULE = "fin"
|
||||
_ENTITIES = [("invoice", "facturas"), ("payment", "pagos")]
|
||||
_ENTITIES = [("invoice", "facturas"), ("payment", "pagos"), ("concept", "conceptos")]
|
||||
_ACTIONS = [("view", "Ver"), ("create", "Crear"), ("edit", "Editar"), ("delete", "Eliminar")]
|
||||
|
||||
|
||||
@@ -12,6 +12,11 @@ def register_permissions() -> None:
|
||||
for entity, label in _ENTITIES:
|
||||
for action, verb in _ACTIONS:
|
||||
registry.register(code=f"{MODULE}.{entity}.{action}", description=f"{verb} {label}", module=MODULE, action=action)
|
||||
# Datos fiscales del emisor: es configuración de la empresa, no una entidad con CRUD,
|
||||
# así que solo tiene ver/editar. Los catálogos del SAT no llevan permiso propio:
|
||||
# son globales y de solo lectura, basta con fin.access.
|
||||
registry.register(code=f"{MODULE}.settings.view", description="Ver datos fiscales del emisor", module=MODULE, action="view")
|
||||
registry.register(code=f"{MODULE}.settings.edit", description="Editar datos fiscales del emisor", module=MODULE, action="edit")
|
||||
|
||||
|
||||
register_permissions()
|
||||
|
||||
@@ -5,8 +5,14 @@ from fastapi import APIRouter, Depends
|
||||
from api.v1.modules.core.permissions.dependencies import PermissionChecker
|
||||
|
||||
from . import permissions # noqa: F401 (side-effect: registra permisos)
|
||||
from .catalogs.routes import router as catalogs_router
|
||||
from .concepts.routes import router as concepts_router
|
||||
from .invoices.routes import router as invoices_router
|
||||
from .issuer.routes import router as issuer_router
|
||||
|
||||
# Enforcement por área/carril (R-T-07): se exige fin.access para el módulo.
|
||||
router = APIRouter(dependencies=[Depends(PermissionChecker(["fin.access"]))])
|
||||
router.include_router(catalogs_router)
|
||||
router.include_router(concepts_router)
|
||||
router.include_router(issuer_router)
|
||||
router.include_router(invoices_router)
|
||||
|
||||
Reference in New Issue
Block a user