Especificación Técnica: cat_grupos_economicos¶
Esquema: clientes
Base de Datos: crm
Tabla Legacy Origen: i143t_grupo_economico (co_grupo_economico, tx_grupo)
Propósito: Permite agrupar diferentes clientes y razones sociales bajo un mismo holding empresarial para consolidar análisis de riesgo y cupo global de crédito.
1. Justificación y Mejoras de Arquitectura¶
-
Estandariza catálogos tipados con UUID, multitenancy mediante RLS por empresa y claves únicas de negocio, eliminando cadenas sueltas y estructuras desnormalizadas.
-
Multi-tenancy RLS y Auditoría Estándar: La tabla incluye
empresa_id BIGINT NOT NULL, utilizado para el aislamiento multiempresa mediante políticas RLS y como identificador de la empresa encore.identidad. Además, incorpora trazabilidad mediantecreated_at,updated_at,deleted_at,created_by,updated_byyversion. -
Unicidad por empresa: El código del grupo económico es único dentro de cada empresa mediante
UNIQUE (empresa_id, codigo). -
Identificación del grupo: El grupo económico representa un holding o conglomerado empresarial al que pueden asociarse múltiples clientes o razones sociales.
2. Definiciones de Implementación¶
Table clientes.cat_grupos_economicos {
id uuid [pk, default: `gen_random_uuid()`, note: 'Identificador único UUID']
empresa_id bigint [not null, note: 'BIGINT de la empresa en core.identidad (RLS)']
codigo varchar(30) [not null, note: 'Código único del grupo dentro de la empresa']
nombre varchar(150) [not null, note: 'Razón social o nombre comercial del holding']
descripcion varchar(255) [note: 'Descripción o detalle del conglomerado']
// 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, codigo) [unique, name: 'uq_cat_grupos_economicos_empresa_codigo']
(empresa_id, is_activo) [name: 'ix_cat_grupos_economicos_empresa_activo']
}
}
CREATE TABLE IF NOT EXISTS clientes.cat_grupos_economicos (
id UUID NOT NULL DEFAULT gen_random_uuid(),
empresa_id BIGINT NOT NULL,
codigo VARCHAR(30) NOT NULL,
nombre VARCHAR(150) NOT NULL,
descripcion VARCHAR(255),
-- Auditoría
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_grupos_economicos
PRIMARY KEY (id),
CONSTRAINT uq_cat_grupos_economicos_empresa_codigo
UNIQUE (empresa_id, codigo)
);
CREATE INDEX IF NOT EXISTS ix_cat_grupos_economicos_empresa_activo
ON clientes.cat_grupos_economicos (
empresa_id,
is_activo
);
ALTER TABLE clientes.cat_grupos_economicos
ENABLE ROW LEVEL SECURITY;
CREATE POLICY rls_cat_grupos_economicos_empresa
ON clientes.cat_grupos_economicos
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
);
COMMENT ON TABLE clientes.cat_grupos_economicos
IS 'Grupos Económicos: Permite agrupar diferentes clientes y razones sociales bajo un mismo holding empresarial para consolidar análisis de riesgo y cupo global de crédito.';
COMMENT ON COLUMN clientes.cat_grupos_economicos.id
IS 'Identificador único UUID';
COMMENT ON COLUMN clientes.cat_grupos_economicos.empresa_id
IS 'Identificador UUID de la empresa en core.identidad (RLS)';
COMMENT ON COLUMN clientes.cat_grupos_economicos.codigo
IS 'Código único del grupo dentro de la empresa';
COMMENT ON COLUMN clientes.cat_grupos_economicos.nombre
IS 'Razón social o nombre comercial del holding';
COMMENT ON COLUMN clientes.cat_grupos_economicos.descripcion
IS 'Descripción o detalle del conglomerado';
COMMENT ON COLUMN clientes.cat_grupos_economicos.is_activo
IS 'Estado lógico del registro';
COMMENT ON COLUMN clientes.cat_grupos_economicos.created_at
IS 'Fecha de creación';
COMMENT ON COLUMN clientes.cat_grupos_economicos.updated_at
IS 'Fecha de última actualización';
COMMENT ON COLUMN clientes.cat_grupos_economicos.deleted_at
IS 'Fecha de eliminación lógica';
COMMENT ON COLUMN clientes.cat_grupos_economicos.created_by
IS 'Usuario creador';
COMMENT ON COLUMN clientes.cat_grupos_economicos.updated_by
IS 'Usuario modificador';
import uuid
from datetime import datetime
from typing import Optional
from sqlalchemy import (
BigInteger,
Boolean,
Index,
String,
UniqueConstraint,
text,
)
from sqlalchemy.dialects.postgresql import TIMESTAMPTZ, UUID
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class CatGruposEconomicos(Base):
"""
Grupos Económicos.
Permite agrupar diferentes clientes y razones sociales
bajo un mismo holding empresarial para consolidar análisis
de riesgo y cupo global de crédito.
Legacy:
i143t_grupo_economico
(co_grupo_economico, tx_grupo_economico)
"""
__tablename__ = "cat_grupos_economicos"
__table_args__ = (
UniqueConstraint(
"empresa_id",
"codigo",
name="uq_cat_grupos_economicos_empresa_codigo",
),
Index(
"ix_cat_grupos_economicos_empresa_activo",
"empresa_id",
"is_activo",
),
{
"schema": "clientes",
"comment": "Grupos Económicos",
},
)
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
primary_key=True,
server_default=text("gen_random_uuid()"),
comment="Identificador único UUID",
)
empresa_id: Mapped[int] = mapped_column(
BigInteger,
nullable=False,
comment="Identificador UUID de la empresa en core.identidad (RLS)",
)
codigo: Mapped[str] = mapped_column(
String(30),
nullable=False,
comment="Código único del grupo dentro de la empresa",
)
nombre: Mapped[str] = mapped_column(
String(150),
nullable=False,
comment="Razón social o nombre comercial del holding",
)
descripcion: Mapped[Optional[str]] = mapped_column(
String(255),
nullable=True,
comment="Descripción o detalle del conglomerado",
)
is_activo: Mapped[bool] = mapped_column(
Boolean,
nullable=False,
server_default=text("true"),
comment="Estado lógico del registro",
)
created_at: Mapped[datetime] = mapped_column(
TIMESTAMPTZ,
nullable=False,
server_default=text("now()"),
comment="Fecha de creación",
)
updated_at: Mapped[datetime] = mapped_column(
TIMESTAMPTZ,
nullable=False,
server_default=text("now()"),
comment="Fecha de última actualización",
)
deleted_at: Mapped[Optional[datetime]] = mapped_column(
TIMESTAMPTZ,
nullable=True,
comment="Fecha de eliminación lógica",
)
created_by: Mapped[Optional[uuid.UUID]] = mapped_column(
UUID(as_uuid=True),
nullable=True,
comment="Usuario creador",
)
updated_by: Mapped[Optional[uuid.UUID]] = mapped_column(
UUID(as_uuid=True),
nullable=True,
comment="Usuario modificador",
)
"""create cat_grupos_economicos
Revision ID: 0004
Revises: 0003
Create Date: 2026-09-09
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision: str = "0004"
down_revision: Union[str, None] = "0003"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"cat_grupos_economicos",
sa.Column(
"id",
postgresql.UUID(as_uuid=True),
nullable=False,
server_default=sa.text("gen_random_uuid()"),
comment="Identificador único UUID",
),
sa.Column(
"empresa_id",
sa.BigInteger(),
nullable=False,
comment="Identificador UUID de la empresa en core.identidad (RLS)",
),
sa.Column(
"codigo",
sa.String(length=30),
nullable=False,
comment="Código único del grupo dentro de la empresa",
),
sa.Column(
"nombre",
sa.String(length=150),
nullable=False,
comment="Razón social o nombre comercial del holding",
),
sa.Column(
"descripcion",
sa.String(length=255),
nullable=True,
comment="Descripción o detalle del conglomerado",
),
# Auditoría
sa.Column(
"is_activo",
sa.Boolean(),
nullable=False,
server_default=sa.text("true"),
comment="Estado lógico del registro",
),
sa.Column(
"created_at",
sa.TIMESTAMP(timezone=True),
nullable=False,
server_default=sa.text("now()"),
comment="Fecha de creación",
),
sa.Column(
"updated_at",
sa.TIMESTAMP(timezone=True),
nullable=False,
server_default=sa.text("now()"),
comment="Fecha de última actualización",
),
sa.Column(
"deleted_at",
sa.TIMESTAMP(timezone=True),
nullable=True,
comment="Fecha de eliminación lógica",
),
sa.Column(
"created_by",
postgresql.UUID(as_uuid=True),
nullable=True,
comment="Usuario creador",
),
sa.Column(
"updated_by",
postgresql.UUID(as_uuid=True),
nullable=True,
comment="Usuario modificador",
),
sa.PrimaryKeyConstraint(
"id",
name="pk_cat_grupos_economicos",
),
sa.UniqueConstraint(
"empresa_id",
"codigo",
name="uq_cat_grupos_economicos_empresa_codigo",
),
schema="clientes",
comment="Grupos Económicos",
)
# Índice
op.create_index(
"ix_cat_grupos_economicos_empresa_activo",
"cat_grupos_economicos",
["empresa_id", "is_activo"],
unique=False,
schema="clientes",
)
# Row Level Security
op.execute(
"""
ALTER TABLE clientes.cat_grupos_economicos
ENABLE ROW LEVEL SECURITY;
"""
)
op.execute(
"""
CREATE POLICY rls_cat_grupos_economicos_empresa
ON clientes.cat_grupos_economicos
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:
# Row Level Security
op.execute(
"""
DROP POLICY IF EXISTS rls_cat_grupos_economicos_empresa
ON clientes.cat_grupos_economicos;
"""
)
op.execute(
"""
ALTER TABLE clientes.cat_grupos_economicos
DISABLE ROW LEVEL SECURITY;
"""
)
# Índice
op.drop_index(
"ix_cat_grupos_economicos_empresa_activo",
table_name="cat_grupos_economicos",
schema="clientes",
)
# Tabla
op.drop_table(
"cat_grupos_economicos",
schema="clientes",
)