Files
evo-sync/web/migrations/versions/0003_rbac_tables.py
mguschin 5ead89e0cf feat: Evotor user lifecycle, RBAC, admin panel
- Receive Evotor webhooks: POST /user/create, /user/verify, /user/token
- Create users in pending status; match to existing users by email/phone
- Send invite link via Celery notification task; user sets password at /invite
- Abstract EmailProvider/SMSProvider with ConsoleEmailProvider default
- Role-based access control: role enum on users + roles/permissions tables
- Admin panel: /admin/users (list, filter, search, paginate), user detail card
  with activate/suspend/reset-password/send-invite/edit/delete actions
- Admin roles management: /admin/roles with per-role permission assignment
- Extend user profile card: role, status, Evotor ID, email confirmation badge
- Auth routes: register, login, logout, confirm-email, forgot/reset password
- Alembic migrations 0002 (full schema + new fields) and 0003 (RBAC + seeds)
- Port Pico CSS + Bootstrap Icons UI from Node.js commit (854c912)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-28 12:01:36 +03:00

106 lines
3.8 KiB
Python

"""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}"))