40 lines
2.9 KiB
Python
40 lines
2.9 KiB
Python
"""Ajouter le RBAC configurable (rôles, permissions et exceptions).
|
|
Revision ID: 9b06c7d8e9f0
|
|
Revises: 8a95b1c2d3e4
|
|
"""
|
|
from alembic import op
|
|
import sqlalchemy as sa
|
|
from app_new.core.models.rbac import Role, Permission, UserRole, RolePermission, UserPermission
|
|
from app_new.core.authorization import PERMISSIONS_BY_ROLE, ROLE_LABELS, GRANULAR_PERMISSION_CODES
|
|
revision = "9b06c7d8e9f0"
|
|
down_revision = "8a95b1c2d3e4"
|
|
branch_labels = None
|
|
depends_on = None
|
|
_TABLES = [Role.__table__, Permission.__table__, UserRole.__table__, RolePermission.__table__, UserPermission.__table__]
|
|
|
|
def upgrade():
|
|
bind = op.get_bind()
|
|
for table in _TABLES:
|
|
table.create(bind=bind, checkfirst=True)
|
|
roles = {slug: ROLE_LABELS.get(slug, slug) for slug in PERMISSIONS_BY_ROLE}
|
|
for slug, name in roles.items():
|
|
bind.execute(sa.text("INSERT INTO roles (slug,name,description,is_system,is_active,created_at,updated_at) SELECT :slug,:name,:description,1,1,NOW(),NOW() FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM roles WHERE slug=:slug)"), {"slug": slug, "name": name, "description": "Rôle système historique"})
|
|
permission_codes = sorted({code for values in PERMISSIONS_BY_ROLE.values() for code in values} | GRANULAR_PERMISSION_CODES)
|
|
for code in permission_codes:
|
|
if code == "*":
|
|
module, action, name = "system", "all", "Accès complet"
|
|
else:
|
|
bits = code.split(".", 1); module = bits[0]; action = bits[1] if len(bits) > 1 else "manage"; name = f"{module.title()} — {action}"
|
|
bind.execute(sa.text("INSERT INTO permissions (code,name,module,action,description,is_active) SELECT :code,:name,:module,:action,:description,1 FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM permissions WHERE code=:code)"), {"code": code, "name": name, "module": module, "action": action, "description": "Permission initiale compatible avec l'ancienne matrice"})
|
|
for slug, codes in PERMISSIONS_BY_ROLE.items():
|
|
role_id = bind.execute(sa.text("SELECT id FROM roles WHERE slug=:slug"), {"slug": slug}).scalar()
|
|
for code in codes:
|
|
permission_id = bind.execute(sa.text("SELECT id FROM permissions WHERE code=:code"), {"code": code}).scalar()
|
|
bind.execute(sa.text("INSERT INTO role_permissions (role_id,permission_id,effect) SELECT :role_id,:permission_id,'allow' FROM DUAL WHERE NOT EXISTS (SELECT 1 FROM role_permissions WHERE role_id=:role_id AND permission_id=:permission_id)"), {"role_id": role_id, "permission_id": permission_id})
|
|
# Rattacher les comptes existants à leur rôle historique sans modifier leur rôle.
|
|
bind.execute(sa.text("INSERT INTO user_roles (user_id,role_id,assigned_at) SELECT u.id,r.id,NOW() FROM users u JOIN roles r ON r.slug=u.role WHERE NOT EXISTS (SELECT 1 FROM user_roles ur WHERE ur.user_id=u.id AND ur.role_id=r.id)"))
|
|
|
|
def downgrade():
|
|
bind = op.get_bind()
|
|
for table in reversed(_TABLES):
|
|
table.drop(bind=bind, checkfirst=True)
|