Especificación Técnica: rel_canal_tipo_empresa¶
Esquema: clientes
Base de Datos: crm
Tabla Legacy Origen: i747t / i748t (configuración de canal y tipo de canal por empresa)
Propósito: Matriz de configuración y habilitación de canales y tipos de canal comercial autorizados para operar en cada tenant empresarial.
1. Justificación y Mejoras de Arquitectura¶
- Parametrización Dinámica por Tenant: Permite activar o desactivar (
is_habilitado) combinaciones específicas de canal comercial y subcanal/tipo de canal por empresa, habilitando modelos comerciales diferenciados entre organizaciones. - Aislamiento Multi-Tenant (RLS): Integra
empresa_id BIGINT NOT NULLcon políticas RLS obligatorias (USINGyWITH CHECK). - Integridad Referencial Compuesta: Clave única compuesta
(empresa_id, id)y claves foráneas compuestas(empresa_id, canal_id)haciaclientes.cat_canales_comerciales(empresa_id, id)y(empresa_id, tipo_canal_id)haciaclientes.cat_tipos_canal_comercial(empresa_id, id), ambas con borrado en cascada (ON DELETE CASCADE). - Unicidad de Configuración: Restricción
UNIQUE (empresa_id, canal_id, tipo_canal_id)para impedir combinaciones duplicadas dentro de la misma empresa. - Trazabilidad y Auditoría: Auditoría integral (
created_at,updated_at,deleted_at,created_by,updated_by) con auditoría integral.
2. Definiciones de Implementación¶
Table clientes.rel_canal_tipo_empresa {
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)']
canal_id uuid [not null, note: 'Canal comercial']
tipo_canal_id uuid [not null, note: 'Subcanal o tipo de canal comercial']
is_habilitado boolean [not null, default: `true`, note: 'Indica si la empresa comercializa en este 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_rel_canal_tipo_empresa_empresa_id']
(empresa_id, canal_id, tipo_canal_id) [unique, name: 'uq_rel_canal_tipo_empresa_asignacion']
(empresa_id, is_activo) [name: 'ix_rel_canal_tipo_empresa_activo']
(empresa_id, canal_id) [name: 'ix_rel_canal_tipo_empresa_canal']
(empresa_id, tipo_canal_id) [name: 'ix_rel_canal_tipo_empresa_tipo']
}
}
Ref: clientes.rel_canal_tipo_empresa.(empresa_id, canal_id) > clientes.cat_canales_comerciales.(empresa_id, id) [delete: cascade]
Ref: clientes.rel_canal_tipo_empresa.(empresa_id, tipo_canal_id) > clientes.cat_tipos_canal_comercial.(empresa_id, id) [delete: cascade]
CREATE TABLE IF NOT EXISTS clientes.rel_canal_tipo_empresa (
id UUID NOT NULL DEFAULT gen_random_uuid(),
empresa_id BIGINT NOT NULL,
canal_id UUID NOT NULL,
tipo_canal_id UUID NOT NULL,
is_habilitado BOOLEAN NOT NULL DEFAULT true,
-- 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_rel_canal_tipo_empresa
PRIMARY KEY (id),
CONSTRAINT uq_rel_canal_tipo_empresa_empresa_id
UNIQUE (empresa_id, id),
CONSTRAINT uq_rel_canal_tipo_empresa_asignacion
UNIQUE (empresa_id, canal_id, tipo_canal_id),
CONSTRAINT fk_rel_canal_tipo_empresa_canal
FOREIGN KEY (empresa_id, canal_id)
REFERENCES clientes.cat_canales_comerciales (empresa_id, id)
ON DELETE CASCADE,
CONSTRAINT fk_rel_canal_tipo_empresa_tipo_canal
FOREIGN KEY (empresa_id, tipo_canal_id)
REFERENCES clientes.cat_tipos_canal_comercial (empresa_id, id)
ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS ix_rel_canal_tipo_empresa_activo
ON clientes.rel_canal_tipo_empresa (empresa_id, is_activo);
CREATE INDEX IF NOT EXISTS ix_rel_canal_tipo_empresa_canal
ON clientes.rel_canal_tipo_empresa (empresa_id, canal_id);
CREATE INDEX IF NOT EXISTS ix_rel_canal_tipo_empresa_tipo
ON clientes.rel_canal_tipo_empresa (empresa_id, tipo_canal_id);
COMMENT ON TABLE clientes.rel_canal_tipo_empresa IS
'Habilitación de Canales Comerciales por Empresa: Configuración de canales y tipos de canal comercial autorizados para operar en cada tenant empresarial.';
COMMENT ON COLUMN clientes.rel_canal_tipo_empresa.id IS
'Identificador único UUID';
COMMENT ON COLUMN clientes.rel_canal_tipo_empresa.empresa_id IS
'Identificador UUID de la empresa en core.identidad (RLS)';
COMMENT ON COLUMN clientes.rel_canal_tipo_empresa.canal_id IS
'Canal comercial';
COMMENT ON COLUMN clientes.rel_canal_tipo_empresa.tipo_canal_id IS
'Subcanal o tipo de canal comercial';
COMMENT ON COLUMN clientes.rel_canal_tipo_empresa.is_habilitado IS
'Indica si la empresa comercializa en este canal';
COMMENT ON COLUMN clientes.rel_canal_tipo_empresa.is_activo IS
'Estado lógico del registro';
COMMENT ON COLUMN clientes.rel_canal_tipo_empresa.created_at IS
'Fecha de creación';
COMMENT ON COLUMN clientes.rel_canal_tipo_empresa.updated_at IS
'Fecha de última actualización';
COMMENT ON COLUMN clientes.rel_canal_tipo_empresa.deleted_at IS
'Fecha de eliminación lógica';
COMMENT ON COLUMN clientes.rel_canal_tipo_empresa.created_by IS
'Usuario creador';
COMMENT ON COLUMN clientes.rel_canal_tipo_empresa.updated_by IS
'Usuario modificador';
ALTER TABLE clientes.rel_canal_tipo_empresa
ENABLE ROW LEVEL SECURITY;
CREATE POLICY rls_rel_canal_tipo_empresa_empresa
ON clientes.rel_canal_tipo_empresa
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,
ForeignKeyConstraint,
Index,
UniqueConstraint,
text,
)
from sqlalchemy.dialects.postgresql import TIMESTAMPTZ, UUID
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base
class RelCanalTipoEmpresa(Base):
"""
Habilitación de Canales Comerciales por Empresa.
Configuración de canales y tipos de canal comercial autorizados
para operar en cada tenant empresarial.
Legacy:
i747t / i748t (configuración de canal y tipo de canal por empresa)
"""
__tablename__ = "rel_canal_tipo_empresa"
__table_args__ = (
UniqueConstraint(
"empresa_id",
"id",
name="uq_rel_canal_tipo_empresa_empresa_id",
),
UniqueConstraint(
"empresa_id",
"canal_id",
"tipo_canal_id",
name="uq_rel_canal_tipo_empresa_asignacion",
),
ForeignKeyConstraint(
["empresa_id", "canal_id"],
["clientes.cat_canales_comerciales.empresa_id", "clientes.cat_canales_comerciales.id"],
ondelete="CASCADE",
name="fk_rel_canal_tipo_empresa_canal",
),
ForeignKeyConstraint(
["empresa_id", "tipo_canal_id"],
["clientes.cat_tipos_canal_comercial.empresa_id", "clientes.cat_tipos_canal_comercial.id"],
ondelete="CASCADE",
name="fk_rel_canal_tipo_empresa_tipo_canal",
),
Index(
"ix_rel_canal_tipo_empresa_activo",
"empresa_id",
"is_activo",
),
Index(
"ix_rel_canal_tipo_empresa_canal",
"empresa_id",
"canal_id",
),
Index(
"ix_rel_canal_tipo_empresa_tipo",
"empresa_id",
"tipo_canal_id",
),
{
"schema": "clientes",
"comment": "Habilitación de Canales Comerciales por Empresa",
},
)
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)",
)
canal_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
nullable=False,
comment="Canal comercial",
)
tipo_canal_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
nullable=False,
comment="Subcanal o tipo de canal comercial",
)
is_habilitado: Mapped[bool] = mapped_column(
Boolean,
nullable=False,
server_default=text("true"),
comment="Indica si la empresa comercializa en este 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",
)
"""create rel_canal_tipo_empresa
Revision ID: rel_0008
Revises: rel_0007
Create Date: 2026-09-10
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision: str = "rel_0008"
down_revision: Union[str, Sequence[str], None] = "rel_0007"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"rel_canal_tipo_empresa",
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(
"canal_id",
postgresql.UUID(as_uuid=True),
nullable=False,
comment="Canal comercial",
),
sa.Column(
"tipo_canal_id",
postgresql.UUID(as_uuid=True),
nullable=False,
comment="Subcanal o tipo de canal comercial",
),
sa.Column(
"is_habilitado",
sa.Boolean(),
nullable=False,
server_default=sa.text("true"),
comment="Indica si la empresa comercializa en este 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",
),
sa.PrimaryKeyConstraint(
"id",
name="pk_rel_canal_tipo_empresa",
),
sa.UniqueConstraint(
"empresa_id",
"id",
name="uq_rel_canal_tipo_empresa_empresa_id",
),
sa.UniqueConstraint(
"empresa_id",
"canal_id",
"tipo_canal_id",
name="uq_rel_canal_tipo_empresa_asignacion",
),
sa.ForeignKeyConstraint(
["empresa_id", "canal_id"],
["clientes.cat_canales_comerciales.empresa_id", "clientes.cat_canales_comerciales.id"],
ondelete="CASCADE",
name="fk_rel_canal_tipo_empresa_canal",
),
sa.ForeignKeyConstraint(
["empresa_id", "tipo_canal_id"],
["clientes.cat_tipos_canal_comercial.empresa_id", "clientes.cat_tipos_canal_comercial.id"],
ondelete="CASCADE",
name="fk_rel_canal_tipo_empresa_tipo_canal",
),
comment="Habilitación de Canales Comerciales por Empresa",
schema="clientes",
)
op.create_index(
"ix_rel_canal_tipo_empresa_activo",
"rel_canal_tipo_empresa",
["empresa_id", "is_activo"],
unique=False,
schema="clientes",
)
op.create_index(
"ix_rel_canal_tipo_empresa_canal",
"rel_canal_tipo_empresa",
["empresa_id", "canal_id"],
unique=False,
schema="clientes",
)
op.create_index(
"ix_rel_canal_tipo_empresa_tipo",
"rel_canal_tipo_empresa",
["empresa_id", "tipo_canal_id"],
unique=False,
schema="clientes",
)
op.execute(
"""
ALTER TABLE clientes.rel_canal_tipo_empresa
ENABLE ROW LEVEL SECURITY;
"""
)
op.execute(
"""
CREATE POLICY rls_rel_canal_tipo_empresa_empresa
ON clientes.rel_canal_tipo_empresa
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_rel_canal_tipo_empresa_empresa
ON clientes.rel_canal_tipo_empresa;
"""
)
op.execute(
"""
ALTER TABLE clientes.rel_canal_tipo_empresa
DISABLE ROW LEVEL SECURITY;
"""
)
op.drop_index(
"ix_rel_canal_tipo_empresa_tipo",
table_name="rel_canal_tipo_empresa",
schema="clientes",
)
op.drop_index(
"ix_rel_canal_tipo_empresa_canal",
table_name="rel_canal_tipo_empresa",
schema="clientes",
)
op.drop_index(
"ix_rel_canal_tipo_empresa_activo",
table_name="rel_canal_tipo_empresa",
schema="clientes",
)
op.drop_table(
"rel_canal_tipo_empresa",
schema="clientes",
)