Especificación Técnica: cat_canales_comerciales¶
Esquema: clientes
Base de Datos: crm
Tabla Legacy Origen: i282t_canales (co_canal, tx_canal)
Propósito: Define los canales de distribución comercial de la empresa (p. ej., Tradicional, Moderno, Horeca, Institucional, Exportación). Permite segmentar estrategias de precios y promociones.
1. Justificación y Mejoras de Arquitectura¶
- Estandariza catálogos tipados con UUID, multitenancy RLS por empresa y claves únicas de código de negocio, eliminando cadenas sueltas y tablas desnormalizadas.
- Multi-tenancy RLS y Auditoría Estándar: Cada tabla incluye
empresa_id BIGINT NOT NULL, utilizado para el aislamiento multiempresa mediante políticas RLS y como identificador de la empresa en core.identidad. Además, incorpora trazabilidad mediantecreated_at,updated_at,deleted_at,created_by,updated_byyversion.
2. Definiciones de Implementación¶
Table clientes.cat_canales_comerciales {
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(30) [not null, note: 'Código único de negocio del canal']
nombre varchar(100) [not null, note: 'Nombre del canal comercial']
descripcion varchar(255) [note: 'Descripción detallada del canal']
// 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_canales_empresa_id']
(empresa_id, codigo) [unique, name: 'uq_cat_canales_empresa_codigo']
(empresa_id, is_activo) [name: 'ix_cat_canales_empresa_activo']
}
}
CREATE TABLE IF NOT EXISTS clientes.cat_canales_comerciales (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
empresa_id BIGINT NOT NULL,
codigo VARCHAR(30) NOT NULL,
nombre VARCHAR(100) NOT NULL,
descripcion VARCHAR(255),
-- 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,
-- Permite FK compuesta desde cat_tipos_canal_comercial
CONSTRAINT uq_cat_canales_empresa_id
UNIQUE (empresa_id, id),
-- Código único dentro de la empresa
CONSTRAINT uq_cat_canales_empresa_codigo
UNIQUE (empresa_id, codigo)
);
CREATE INDEX IF NOT EXISTS ix_cat_canales_empresa_activo
ON clientes.cat_canales_comerciales (
empresa_id,
is_activo
);
COMMENT ON TABLE clientes.cat_canales_comerciales IS
'Canales Comerciales: Define los canales de distribución comercial de la empresa (p. ej., Tradicional, Moderno, Horeca, Institucional, Exportación). Permite segmentar estrategias de precios y promociones.';
COMMENT ON COLUMN clientes.cat_canales_comerciales.id IS
'Identificador único UUID';
COMMENT ON COLUMN clientes.cat_canales_comerciales.empresa_id IS
'Identificador UUID de la empresa en core.identidad (RLS)';
COMMENT ON COLUMN clientes.cat_canales_comerciales.codigo IS
'Código único de negocio del canal';
COMMENT ON COLUMN clientes.cat_canales_comerciales.nombre IS
'Nombre del canal comercial';
COMMENT ON COLUMN clientes.cat_canales_comerciales.descripcion IS
'Descripción detallada del canal';
COMMENT ON COLUMN clientes.cat_canales_comerciales.is_activo IS
'Estado lógico del registro';
COMMENT ON COLUMN clientes.cat_canales_comerciales.created_at IS
'Fecha de creación';
COMMENT ON COLUMN clientes.cat_canales_comerciales.updated_at IS
'Fecha de última actualización';
COMMENT ON COLUMN clientes.cat_canales_comerciales.deleted_at IS
'Fecha de eliminación lógica';
COMMENT ON COLUMN clientes.cat_canales_comerciales.created_by IS
'Usuario creador';
COMMENT ON COLUMN clientes.cat_canales_comerciales.updated_by IS
'Usuario modificador';
ALTER TABLE clientes.cat_canales_comerciales
ENABLE ROW LEVEL SECURITY;
CREATE POLICY rls_cat_canales_comerciales_empresa
ON clientes.cat_canales_comerciales
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
);
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 (
DeclarativeBase,
Mapped,
mapped_column,
)
class Base(DeclarativeBase):
pass
class CatCanalesComerciales(Base):
"""
Canales Comerciales.
Define los canales de distribución comercial de la empresa
(p. ej., Tradicional, Moderno, Horeca, Institucional, Exportación).
Legacy:
i282t_canales (co_canal, tx_canal)
"""
__tablename__ = "cat_canales_comerciales"
__table_args__ = (
UniqueConstraint(
"empresa_id",
"id",
name="uq_cat_canales_empresa_id",
),
UniqueConstraint(
"empresa_id",
"codigo",
name="uq_cat_canales_empresa_codigo",
),
Index(
"ix_cat_canales_empresa_activo",
"empresa_id",
"is_activo",
),
{
"schema": "clientes",
"comment": "Canales Comerciales",
},
)
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 de negocio del canal",
)
nombre: Mapped[str] = mapped_column(
String(100),
nullable=False,
comment="Nombre del canal comercial",
)
descripcion: Mapped[Optional[str]] = mapped_column(
String(255),
nullable=True,
comment="Descripción detallada del canal",
)
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",
)
``
`python """create cat_canales_comerciales
Revision ID: 0001 Revises: Create Date: 2026-09-09 """
from typing import Sequence, Union
from alembic import op import sqlalchemy as sa from sqlalchemy.dialects import postgresql
============================================================¶
Revision identifiers¶
============================================================¶
revision: str = "0001" down_revision: Union[str, Sequence[str], None] = None branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ========================================================
# Tabla
# ========================================================
op.create_table(
"cat_canales_comerciales",
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(30),
nullable=False,
comment="Código único de negocio del canal",
),
sa.Column(
"nombre",
sa.String(100),
nullable=False,
comment="Nombre del canal comercial",
),
sa.Column(
"descripcion",
sa.String(255),
nullable=True,
comment="Descripción detallada del canal",
),
# Auditoría General
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",
),
# ====================================================
# Constraints
# ====================================================
sa.PrimaryKeyConstraint(
"id",
name="pk_cat_canales_comerciales",
),
# Necesario para FK compuesta de cat_tipos_canal_comercial
sa.UniqueConstraint(
"empresa_id",
"id",
name="uq_cat_canales_empresa_id",
),
# Código único por empresa
sa.UniqueConstraint(
"empresa_id",
"codigo",
name="uq_cat_canales_empresa_codigo",
),
schema="clientes",
comment="Canales Comerciales",
)
# ========================================================
# Índices
# ========================================================
op.create_index(
"ix_cat_canales_empresa_activo",
"cat_canales_comerciales",
["empresa_id", "is_activo"],
unique=False,
schema="clientes",
)
# ========================================================
# Row Level Security
# ========================================================
op.execute(
"""
ALTER TABLE clientes.cat_canales_comerciales
ENABLE ROW LEVEL SECURITY
"""
)
# ========================================================
# RLS Policy
# ========================================================
op.execute(
"""
CREATE POLICY rls_cat_canales_comerciales_empresa
ON clientes.cat_canales_comerciales
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:
# ========================================================
# Eliminar RLS Policy
# ========================================================
op.execute(
"""
DROP POLICY IF EXISTS
rls_cat_canales_comerciales_empresa
ON clientes.cat_canales_comerciales
"""
)
# ========================================================
# Deshabilitar RLS
# ========================================================
op.execute(
"""
ALTER TABLE clientes.cat_canales_comerciales
DISABLE ROW LEVEL SECURITY
"""
)
# ========================================================
# Eliminar índice
# ========================================================
op.drop_index(
"ix_cat_canales_empresa_activo",
table_name="cat_canales_comerciales",
schema="clientes",
)
# ========================================================
# Eliminar tabla
# ========================================================
op.drop_table(
"cat_canales_comerciales",
schema="clientes",
)
```