leocrm/alembic/versions/0049_entity_permissions.py

"""Universal entity_permissions table — ACLs for ALL entities.

Revision ID: 0049
Revises: 0048
Create Date: 2026-07-29
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID as PGUUID

revision = "0049"
down_revision = "0048"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table(
        "entity_permissions",
        sa.Column("id", PGUUID(as_uuid=True), primary_key=True),
        sa.Column("entity_type", sa.String(50), nullable=False),
        sa.Column("entity_id", PGUUID(as_uuid=True), nullable=False),
        sa.Column("principal_type", sa.String(10), nullable=False),
        sa.Column("principal_id", PGUUID(as_uuid=True), nullable=False),
        sa.Column("permission_level", sa.String(20), nullable=False, server_default="read"),
        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("created_by", PGUUID(as_uuid=True), sa.ForeignKey("users.id", ondelete="SET NULL"), nullable=True),
        sa.Column("tenant_id", PGUUID(as_uuid=True), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=False),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
        sa.UniqueConstraint("entity_type", "entity_id", "principal_type", "principal_id", "tenant_id", name="uq_ep_entity_principal_tenant"),
        sa.CheckConstraint("principal_type IN ('user', 'group', 'role', 'guest')", name="ck_ep_principal_type"),
        sa.CheckConstraint("permission_level IN ('none', 'read', 'write', 'admin', 'delete')", name="ck_ep_permission_level"),
    )
    op.execute('CREATE INDEX IF NOT EXISTS ix_ep_entity ON entity_permissions (entity_type, entity_id)')
    op.execute('CREATE INDEX IF NOT EXISTS ix_ep_principal ON entity_permissions (principal_type, principal_id)')
    op.execute('CREATE INDEX IF NOT EXISTS ix_ep_tenant ON entity_permissions (tenant_id)')
    op.execute('CREATE INDEX IF NOT EXISTS ix_ep_expires ON entity_permissions (expires_at)')


def downgrade() -> None:
    op.drop_index("ix_ep_expires", table_name="entity_permissions")
    op.drop_index("ix_ep_tenant", table_name="entity_permissions")
    op.drop_index("ix_ep_principal", table_name="entity_permissions")
    op.drop_index("ix_ep_entity", table_name="entity_permissions")
    op.drop_table("entity_permissions")