evo-sync/web/migrations/versions/0003_rbac_tables.py

"""RBAC tables with default roles and permissions

Revision ID: 0003
Revises: 0002
Create Date: 2026-04-28

"""
from typing import Sequence, Union

from alembic import op
import sqlalchemy as sa

revision: str = "0003"
down_revision: Union[str, None] = "0002"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None

DEFAULT_ROLES = [
    ("system", "Системный администратор — полный доступ"),
    ("admin",  "Администратор — управление пользователями"),
    ("user",   "Обычный пользователь"),
]

DEFAULT_PERMISSIONS = [
    ("admin.users.view",    "Просмотр списка пользователей"),
    ("admin.users.edit",    "Редактирование пользователей"),
    ("admin.users.delete",  "Удаление пользователей"),
    ("admin.roles.manage",  "Управление ролями и правами"),
]

# system gets all permissions; admin gets view+edit
ROLE_PERMISSION_MAP = {
    "system": ["admin.users.view", "admin.users.edit", "admin.users.delete", "admin.roles.manage"],
    "admin":  ["admin.users.view", "admin.users.edit"],
    "user":   [],
}


def upgrade() -> None:
    conn = op.get_bind()

    conn.execute(sa.text("""
        CREATE TABLE IF NOT EXISTS roles (
            id          INT AUTO_INCREMENT PRIMARY KEY,
            name        VARCHAR(50) NOT NULL,
            description VARCHAR(255) NULL,
            CONSTRAINT uq_roles_name UNIQUE (name)
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci
    """))

    conn.execute(sa.text("""
        CREATE TABLE IF NOT EXISTS permissions (
            id          INT AUTO_INCREMENT PRIMARY KEY,
            name        VARCHAR(100) NOT NULL,
            description VARCHAR(255) NULL,
            CONSTRAINT uq_permissions_name UNIQUE (name)
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci
    """))

    conn.execute(sa.text("""
        CREATE TABLE IF NOT EXISTS role_permissions (
            role_id       INT NOT NULL,
            permission_id INT NOT NULL,
            PRIMARY KEY (role_id, permission_id),
            FOREIGN KEY (role_id) REFERENCES roles(id) ON DELETE CASCADE,
            FOREIGN KEY (permission_id) REFERENCES permissions(id) ON DELETE CASCADE
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci
    """))

    conn.execute(sa.text("""
        CREATE TABLE IF NOT EXISTS user_roles (
            user_id INT NOT NULL,
            role_id INT NOT NULL,
            PRIMARY KEY (user_id, role_id),
            FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
            FOREIGN KEY (role_id) REFERENCES roles(id) ON DELETE CASCADE
        ) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci
    """))

    # Seed default roles
    for name, description in DEFAULT_ROLES:
        conn.execute(sa.text(
            "INSERT IGNORE INTO roles (name, description) VALUES (:name, :desc)"
        ), {"name": name, "desc": description})

    # Seed default permissions
    for name, description in DEFAULT_PERMISSIONS:
        conn.execute(sa.text(
            "INSERT IGNORE INTO permissions (name, description) VALUES (:name, :desc)"
        ), {"name": name, "desc": description})

    # Seed role_permissions
    for role_name, perm_names in ROLE_PERMISSION_MAP.items():
        for perm_name in perm_names:
            conn.execute(sa.text("""
                INSERT IGNORE INTO role_permissions (role_id, permission_id)
                SELECT r.id, p.id FROM roles r, permissions p
                WHERE r.name = :role AND p.name = :perm
            """), {"role": role_name, "perm": perm_name})


def downgrade() -> None:
    conn = op.get_bind()
    for table in ["user_roles", "role_permissions", "permissions", "roles"]:
        conn.execute(sa.text(f"DROP TABLE IF EXISTS {table}"))