63 lines
2 KiB
Python
63 lines
2 KiB
Python
import logging
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.rbac import Permission, Role
|
|
from app.rbac.defaults import ROLE_PERMISSION_NAMES, STANDARD_PERMISSIONS, STANDARD_ROLES
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def seed_rbac(db: Session) -> None:
|
|
roles_by_name = {
|
|
role.name: role
|
|
for role in db.scalars(select(Role)).all()
|
|
}
|
|
permissions_by_name = {
|
|
permission.name: permission
|
|
for permission in db.scalars(select(Permission)).all()
|
|
}
|
|
|
|
for role_data in STANDARD_ROLES:
|
|
role = roles_by_name.get(role_data["name"])
|
|
if role is None:
|
|
role = Role(**role_data)
|
|
db.add(role)
|
|
roles_by_name[role.name] = role
|
|
else:
|
|
role.display_name = role_data["display_name"]
|
|
role.description = role_data["description"]
|
|
role.is_system = role_data["is_system"]
|
|
|
|
for name, display_name, description, module in STANDARD_PERMISSIONS:
|
|
permission = permissions_by_name.get(name)
|
|
if permission is None:
|
|
permission = Permission(
|
|
name=name,
|
|
display_name=display_name,
|
|
description=description,
|
|
module=module,
|
|
)
|
|
db.add(permission)
|
|
permissions_by_name[name] = permission
|
|
else:
|
|
permission.display_name = display_name
|
|
permission.description = description
|
|
permission.module = module
|
|
|
|
db.flush()
|
|
|
|
all_permissions = set(permissions_by_name.values())
|
|
for role_name, permission_names in ROLE_PERMISSION_NAMES.items():
|
|
role = roles_by_name[role_name]
|
|
if permission_names == "*":
|
|
role.permissions = list(all_permissions)
|
|
else:
|
|
role.permissions = [
|
|
permissions_by_name[permission_name]
|
|
for permission_name in permission_names
|
|
]
|
|
|
|
db.commit()
|
|
logger.info("rbac.seeded")
|