Especificación Técnica: cfg_registro_cliente_sucursal¶
Esquema: clientes
Base de Datos: crm
Servicio: svc-clientes
Tabla Legacy Origen: i406t (defaults)
Propósito: valores por defecto del formulario de alta, por sucursal
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 clientes.cfg_registro_cliente_sucursal {
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)']
sucursal_id uuid [not null, note: 'Correlación lógica con core.identidad.mae_sucursales (sin FK cross-DB)']
magnitud_empresa_id_default uuid [, note: 'Campo magnitud_empresa_id_default']
tipo_canal_id_default uuid [, note: 'Campo tipo_canal_id_default']
tipo_credito_id_default uuid [, note: 'Campo tipo_credito_id_default']
tipo_envio_id_default uuid [, note: 'Campo tipo_envio_id_default']
tipo_cliente_id_default uuid [, note: 'Campo tipo_cliente_id_default']
clase_cliente_id_default uuid [, note: 'Campo clase_cliente_id_default']
canal_id_default uuid [, note: 'Campo canal_id_default']
documento_legal_erp_id_default uuid [, note: 'Campo documento_legal_erp_id_default']
activa_filtro_credito boolean [not null, note: 'Campo activa_filtro_credito']
// 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_cfg_registro_cliente_sucursal_empresa_id']
(empresa_id, sucursal_id) [name: 'ix_cfg_registro_cliente_sucursal_empresa_sucursal']
(empresa_id, is_activo) [name: 'ix_cfg_registro_cliente_sucursal_empresa_activo']
}
}
CREATE TABLE IF NOT EXISTS clientes.cfg_registro_cliente_sucursal (
id UUID NOT NULL DEFAULT gen_random_uuid(),
empresa_id BIGINT NOT NULL,
sucursal_id UUID NOT NULL,
magnitud_empresa_id_default UUID,
tipo_canal_id_default UUID,
tipo_credito_id_default UUID,
tipo_envio_id_default UUID,
tipo_cliente_id_default UUID,
clase_cliente_id_default UUID,
canal_id_default UUID,
documento_legal_erp_id_default UUID,
activa_filtro_credito BOOLEAN 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_cfg_registro_cliente_sucursal PRIMARY KEY (id),
CONSTRAINT uq_cfg_registro_cliente_sucursal_empresa_id UNIQUE (empresa_id, id)
);
CREATE INDEX IF NOT EXISTS ix_cfg_registro_cliente_sucursal_empresa_sucursal ON clientes.cfg_registro_cliente_sucursal (empresa_id, sucursal_id);
CREATE INDEX IF NOT EXISTS ix_cfg_registro_cliente_sucursal_empresa_activo ON clientes.cfg_registro_cliente_sucursal (empresa_id, is_activo);
ALTER TABLE clientes.cfg_registro_cliente_sucursal ENABLE ROW LEVEL SECURITY;
CREATE POLICY rls_cfg_registro_cliente_sucursal_tenant_isolation ON clientes.cfg_registro_cliente_sucursal
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 RegistroClienteSucursal(Base):
__tablename__ = 'cfg_registro_cliente_sucursal'
__table_args__ = (
UniqueConstraint('empresa_id', 'id', name='uq_cfg_registro_cliente_sucursal_empresa_id'),
Index('ix_cfg_registro_cliente_sucursal_empresa_sucursal', 'empresa_id', 'sucursal_id'),
Index('ix_cfg_registro_cliente_sucursal_empresa_activo', 'empresa_id', 'is_activo'),
{'schema': 'clientes'}
)
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)
sucursal_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False)
magnitud_empresa_id_default: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)
tipo_canal_id_default: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)
tipo_credito_id_default: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)
tipo_envio_id_default: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)
tipo_cliente_id_default: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)
clase_cliente_id_default: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)
canal_id_default: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)
documento_legal_erp_id_default: Mapped[Optional[uuid.UUID]] = mapped_column(UUID(as_uuid=True), nullable=True)
activa_filtro_credito: Mapped[bool] = mapped_column(Boolean, default=True, 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: crm_cli_0021
create table clientes.cfg_registro_cliente_sucursal
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision = 'crm_cli_0021'
down_revision = 'crm_cli_0020'
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
'cfg_registro_cliente_sucursal',
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('sucursal_id', postgresql.UUID(as_uuid=True), nullable=False),
sa.Column('magnitud_empresa_id_default', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('tipo_canal_id_default', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('tipo_credito_id_default', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('tipo_envio_id_default', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('tipo_cliente_id_default', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('clase_cliente_id_default', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('canal_id_default', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('documento_legal_erp_id_default', postgresql.UUID(as_uuid=True), nullable=True),
sa.Column('activa_filtro_credito', sa.Boolean(), 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_cfg_registro_cliente_sucursal'),
sa.UniqueConstraint('empresa_id', 'id', name='uq_cfg_registro_cliente_sucursal_empresa_id'),
schema='clientes'
)
op.create_index('ix_cfg_registro_cliente_sucursal_empresa_sucursal', 'cfg_registro_cliente_sucursal', ['empresa_id', 'sucursal_id'], unique=False, schema='clientes')
op.create_index('ix_cfg_registro_cliente_sucursal_empresa_activo', 'cfg_registro_cliente_sucursal', ['empresa_id', 'is_activo'], unique=False, schema='clientes')
op.execute('ALTER TABLE clientes.cfg_registro_cliente_sucursal ENABLE ROW LEVEL SECURITY')
op.execute('''CREATE POLICY rls_cfg_registro_cliente_sucursal_tenant_isolation ON clientes.cfg_registro_cliente_sucursal 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_cfg_registro_cliente_sucursal_tenant_isolation ON clientes.cfg_registro_cliente_sucursal')
op.drop_index('ix_cfg_registro_cliente_sucursal_empresa_sucursal', table_name='cfg_registro_cliente_sucursal', schema='clientes')
op.drop_index('ix_cfg_registro_cliente_sucursal_empresa_activo', table_name='cfg_registro_cliente_sucursal', schema='clientes')
op.drop_table('cfg_registro_cliente_sucursal', schema='clientes')