Especificación Técnica: cat_bancos¶
Esquema: contable
Base de Datos: erp
Servicio: svc-contable
Tabla Legacy Origen: co_banco
Propósito: catálogos de cobranza de campo
1. Justificación y Mejoras de Arquitectura¶
- Normalización Relacional Rigurosa: Aplica diseño normalizado (1FN a BCNF), eliminando grupos repetidos de booleanos, desnormalizaciones innecesarias y redundancias del sistema legacy.
- Aislamiento Multi-Tenant (RLS): Integra
empresa_id BIGINT NOT NULLindexado, gobernado por políticas PostgreSQL Row-Level Security (USINGyWITH CHECK) para garantizar la total separación de datos por tenant. - Integridad Referencial Compuesta: Todas las claves foráneas internas hacia tablas con tenencia son compuestas
(empresa_id, <fk_id>), asegurando la integridad física en PostgreSQL. - Trazabilidad y Auditoría: Auditoría estándar (
created_at,updated_at,deleted_at,created_by,updated_by) con auditoría integral.
2. Definiciones de Implementación¶
Table contable.cat_bancos {
id uuid [pk, default: `gen_random_uuid()`, note: 'Identificador único UUID']
empresa_id bigint [not null, note: 'Identificador BIGINT de la empresa en core.identidad (RLS)']
codigo varchar(20) [not null, note: 'Campo codigo']
nombre varchar(100) [not null, note: 'Campo nombre']
// Auditoría General
is_activo boolean [not null, default: `true`, note: 'Estado lógico del registro']
created_at timestamptz [not null, default: `now()`, note: 'Fecha de creación']
updated_at timestamptz [not null, default: `now()`, note: 'Fecha de última actualización']
deleted_at timestamptz [note: 'Fecha de eliminación lógica']
created_by uuid [note: 'Usuario creador']
updated_by uuid [note: 'Usuario modificador']
Indexes {
(empresa_id, id) [unique, name: 'uq_cat_bancos_empresa_id']
(empresa_id, codigo) [unique, name: 'uq_cat_bancos_empresa_codigo']
(empresa_id, is_activo) [name: 'ix_cat_bancos_empresa_activo']
}
}
CREATE TABLE IF NOT EXISTS contable.cat_bancos (
id UUID NOT NULL DEFAULT gen_random_uuid(),
empresa_id BIGINT NOT NULL,
codigo VARCHAR(20) NOT NULL,
nombre VARCHAR(100) NOT NULL,
-- Auditoría General
is_activo BOOLEAN NOT NULL DEFAULT true,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
deleted_at TIMESTAMPTZ,
created_by UUID,
updated_by UUID,
CONSTRAINT pk_cat_bancos PRIMARY KEY (id),
CONSTRAINT uq_cat_bancos_empresa_id UNIQUE (empresa_id, id),
CONSTRAINT uq_cat_bancos_empresa_codigo UNIQUE (empresa_id, codigo)
);
CREATE INDEX IF NOT EXISTS ix_cat_bancos_empresa_activo ON contable.cat_bancos (empresa_id, is_activo);
ALTER TABLE contable.cat_bancos ENABLE ROW LEVEL SECURITY;
CREATE POLICY rls_cat_bancos_tenant_isolation ON contable.cat_bancos
FOR ALL
USING (empresa_id = current_setting('app.current_empresa_id', true)::BIGINT)
WITH CHECK (empresa_id = current_setting('app.current_empresa_id', true)::BIGINT);
from datetime import datetime
import uuid
from typing import Optional
from sqlalchemy import String, Boolean, DateTime, BigInteger, ForeignKey, ForeignKeyConstraint, UniqueConstraint, Index, Numeric, Integer, Text
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase
class Base(DeclarativeBase):
pass
class Bancos(Base):
__tablename__ = 'cat_bancos'
__table_args__ = (
UniqueConstraint('empresa_id', 'id', name='uq_cat_bancos_empresa_id'),
UniqueConstraint('empresa_id', 'codigo', name='uq_cat_bancos_empresa_codigo'),
Index('ix_cat_bancos_empresa_activo', 'empresa_id', 'is_activo'),
{'schema': 'contable'}
)
id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
empresa_id: Mapped[int] = mapped_column(
BigInteger, nullable=False)
codigo: Mapped[str] = mapped_column(String(200), nullable=False)
nombre: Mapped[str] = mapped_column(String(200), nullable=False)
# Auditoría
is_activo: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=datetime.utcnow, nullable=False)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
deleted_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
created_by: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)
updated_by: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)
"""revision: erp_con_0001
create table contable.cat_bancos
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision = 'erp_con_0001'
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
'cat_bancos',
sa.Column('id', postgresql.UUID(as_uuid=True), server_default=sa.text('gen_random_uuid()'), nullable=False),
sa.Column(
"empresa_id",
sa.BigInteger(), nullable=False),
sa.Column('codigo', sa.String(length=200), nullable=False),
sa.Column('nombre', sa.String(length=200), nullable=False),
sa.Column('is_activo', sa.Boolean(), server_default=sa.text('true'), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
sa.Column('deleted_at', sa.DateTime(timezone=True), nullable=True),
sa.Column('created_by', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('updated_by', postgresql.UUID(as_uuid=True), nullable=True),
sa.PrimaryKeyConstraint('id', name='pk_cat_bancos'),
sa.UniqueConstraint('empresa_id', 'id', name='uq_cat_bancos_empresa_id'),
sa.UniqueConstraint('empresa_id', 'codigo', name='uq_cat_bancos_empresa_codigo'),
UniqueConstraint('empresa_id', 'codigo', name='uq_cat_bancos_empresa_codigo'),
schema='contable'
)
op.create_index('ix_cat_bancos_empresa_activo', 'cat_bancos', ['empresa_id', 'is_activo'], unique=False, schema='contable')
op.execute('ALTER TABLE contable.cat_bancos ENABLE ROW LEVEL SECURITY')
op.execute('''CREATE POLICY rls_cat_bancos_tenant_isolation ON contable.cat_bancos FOR ALL USING (empresa_id = current_setting(\'app.current_empresa_id\', true)::BIGINT) WITH CHECK (empresa_id = current_setting(\'app.current_empresa_id\', true)::BIGINT)''')
def downgrade() -> None:
op.execute('DROP POLICY IF EXISTS rls_cat_bancos_tenant_isolation ON contable.cat_bancos')
op.drop_index('ix_cat_bancos_empresa_activo', table_name='cat_bancos', schema='contable')
op.drop_table('cat_bancos', schema='contable')