Especificación Técnica: rel_usuario_empresa¶
Esquema: core.identidad
Base de Datos: core
Servicio: svc-identidad
Propósito: Relación N:M entre usuarios y empresas a las que pueden acceder.
1. Justificación y Mejoras de Arquitectura¶
- Define la membresía de un usuario a una o varias empresas.
- Unique (usuario_id, empresa_id) impide membresías duplicadas.
- Tabla raíz de acceso: sin RLS porque no pertenece a una única empresa.
2. Definiciones de Implementación¶
Table core.identidad.rel_usuario_empresa {
id uuid [pk, default: `gen_random_uuid()`]
usuario_id uuid [not null, ref: > mae_usuarios.id]
empresa_id bigint [not null, ref: > mae_empresas.id]
// 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 {
(usuario_id, empresa_id) [unique, name: 'uq_rel_usuario_empresa']
}
}
CREATE TABLE IF NOT EXISTS core.identidad.rel_usuario_empresa (
id UUID NOT NULL DEFAULT gen_random_uuid() PRIMARY KEY,
usuario_id UUID NOT NULL,
empresa_id BIGINT 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 uq_rel_usuario_empresa UNIQUE (usuario_id, empresa_id),
CONSTRAINT fk_rel_usuario_empresa_usuario FOREIGN KEY (usuario_id) REFERENCES core.identidad.mae_usuarios (id),
CONSTRAINT fk_rel_usuario_empresa_empresa FOREIGN KEY (empresa_id) REFERENCES core.identidad.mae_empresas (id)
);
-- Comentarios
COMMENT ON COLUMN core.identidad.rel_usuario_empresa.usuario_id IS 'Usuario';
COMMENT ON COLUMN core.identidad.rel_usuario_empresa.empresa_id IS 'Empresa';
from datetime import datetime, date
from typing import Optional
import uuid
from sqlalchemy import (
BigInteger,
Boolean,
CheckConstraint,
Date,
DateTime,
ForeignKey,
Identity,
Index,
Integer,
SmallInteger,
String,
Text,
UniqueConstraint,
text,
)
from sqlalchemy.dialects import postgresql
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase):
pass
class RelUsuarioEmpresa(Base):
__tablename__ = 'rel_usuario_empresa'
__table_args__ = (
UniqueConstraint('usuario_id', 'empresa_id', name='uq_rel_usuario_empresa'),
{"schema": 'core.identidad'},
)
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
primary_key=True,
server_default=text("gen_random_uuid()"),
)
usuario_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
nullable=False,
ForeignKey("mae_usuarios.id"),
comment='Usuario',
)
empresa_id: Mapped[int] = mapped_column(
BigInteger,
nullable=False,
ForeignKey("mae_empresas.id"),
comment='Empresa',
)
# Auditoría General
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(
DateTime(timezone=True),
nullable=False,
server_default=text("now()"),
comment='Fecha de creación',
)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
server_default=text("now()"),
comment='Fecha de última actualización',
)
deleted_at: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True),
comment='Fecha de eliminación lógica',
)
created_by: Mapped[uuid.UUID | None] = mapped_column(
UUID(as_uuid=True),
comment='Usuario creador',
)
updated_by: Mapped[uuid.UUID | None] = mapped_column(
UUID(as_uuid=True),
comment='Usuario modificador',
)
"""revision: core_id_0018
create table core.identidad.rel_usuario_empresa
"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
revision: str = 'core_id_0018'
down_revision: str | None = 'core_id_0017'
branch_labels: str | None = None
depends_on: str | None = None
def upgrade() -> None:
op.create_table(
'rel_usuario_empresa',
sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")),
sa.Column("usuario_id", postgresql.UUID(as_uuid=True), nullable=False),
sa.Column("empresa_id", sa.BigInteger(), nullable=False),
sa.Column("is_activo", sa.Boolean(), nullable=False, server_default=sa.text("true")),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
sa.Column("deleted_at", sa.DateTime(timezone=True)),
sa.Column("created_by", postgresql.UUID(as_uuid=True)),
sa.Column("updated_by", postgresql.UUID(as_uuid=True)),
sa.UniqueConstraint('usuario_id', 'empresa_id', name='uq_rel_usuario_empresa'),
sa.ForeignKeyConstraint(['usuario_id'], ['core.identidad.mae_usuarios.id'], name='fk_rel_usuario_empresa_usuario'),
sa.ForeignKeyConstraint(['empresa_id'], ['core.identidad.mae_empresas.id'], name='fk_rel_usuario_empresa_empresa'),
schema='core.identidad',
)
def downgrade() -> None:
op.drop_table('rel_usuario_empresa', schema='core.identidad')