Agent Zero
48 lines
2.4 KiB
Python
48 lines
2.4 KiB
Python
"""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")
|