Files

165 lines
5.6 KiB
Python

"""Idempotent seed for roles, permissions, and the Hunter Stock Feeds users.
Re-running this is safe: it upserts permissions, syncs each role's permission
set to the declared list, and creates or updates the seed users without
duplicating rows. Permission grants are the source of truth — change them
here (or in the DB) rather than in route code.
"""
from __future__ import annotations
from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload
from app.models.access import Permission, Role, User
PERMISSION_DEFINITIONS: tuple[tuple[str, str], ...] = (
("view_dashboard", "View the operational dashboard"),
("view_mix_calculator", "View the mix calculator area"),
("use_mix_calculator", "Run calculations in the mix calculator"),
("save_mix_calculator_session", "Save mix calculator sessions"),
("view_raw_materials", "View raw materials"),
("edit_raw_materials", "Create and edit raw materials"),
("view_products", "View finished products"),
("edit_products", "Create and edit finished products"),
("view_mixes", "View mix master recipes"),
("edit_mixes", "Create and edit mix master recipes"),
("view_users", "View internal users and roles"),
("manage_users", "Create, deactivate, and assign user roles"),
("manage_permissions", "Modify roles and role-permission assignments"),
("view_settings", "View system settings"),
("edit_settings", "Edit system settings"),
)
ROLE_DEFINITIONS: dict[str, dict] = {
"Admin": {
"description": "Full administrative access including user and permission management.",
"permissions": [
"view_dashboard",
"view_mix_calculator",
"use_mix_calculator",
"save_mix_calculator_session",
"view_raw_materials",
"edit_raw_materials",
"view_products",
"view_mixes",
"view_users",
"manage_users",
"manage_permissions",
"view_settings",
"edit_settings",
],
},
"Operations": {
"description": "Mix calculator only — cannot edit raw materials, products, mixes, users, or settings.",
"permissions": [
"view_mix_calculator",
"use_mix_calculator",
"save_mix_calculator_session",
],
},
"Full Access": {
"description": "Operational data editor — cannot manage users or permissions unless explicitly granted.",
"permissions": [
"view_dashboard",
"view_mix_calculator",
"use_mix_calculator",
"save_mix_calculator_session",
"view_raw_materials",
"edit_raw_materials",
"view_products",
"edit_products",
"view_mixes",
"edit_mixes",
],
},
}
SEED_USERS: tuple[dict, ...] = (
{
"email": "admin@hunterstockfeeds.com",
"name": "Hunter Stock Feeds Admin",
"role": "Admin",
},
{
"email": "ops@hunterstockfeeds.com",
"name": "Hunter Stock Feeds Operations",
"role": "Operations",
},
{
"email": "craig@hunterstockfeeds.com",
"name": "Craig",
"role": "Full Access",
},
)
def _upsert_permissions(db: Session) -> dict[str, Permission]:
existing = {permission.key: permission for permission in db.scalars(select(Permission)).all()}
for key, description in PERMISSION_DEFINITIONS:
permission = existing.get(key)
if permission is None:
permission = Permission(key=key, description=description)
db.add(permission)
existing[key] = permission
elif permission.description != description:
permission.description = description
db.flush()
return existing
def _upsert_roles(db: Session, permissions_by_key: dict[str, Permission]) -> dict[str, Role]:
existing = {
role.name: role
for role in db.scalars(
select(Role).options(selectinload(Role.permissions))
).all()
}
for role_name, definition in ROLE_DEFINITIONS.items():
role = existing.get(role_name)
if role is None:
role = Role(name=role_name, description=definition["description"])
db.add(role)
existing[role_name] = role
elif role.description != definition["description"]:
role.description = definition["description"]
db.flush()
desired = {permissions_by_key[key] for key in definition["permissions"]}
current = set(role.permissions)
for permission in desired - current:
role.permissions.append(permission)
for permission in current - desired:
role.permissions.remove(permission)
db.flush()
return existing
def _upsert_users(db: Session, roles_by_name: dict[str, Role]) -> None:
existing = {user.email: user for user in db.scalars(select(User)).all()}
for entry in SEED_USERS:
email = entry["email"].lower()
role = roles_by_name[entry["role"]]
user = existing.get(email)
if user is None:
user = User(email=email, name=entry["name"], role_id=role.id, is_active=True)
db.add(user)
existing[email] = user
else:
user.name = entry["name"]
user.role_id = role.id
if not user.is_active:
user.is_active = True
db.flush()
def seed_access(db: Session) -> None:
"""Idempotent: roles, permissions, role-permission links, seed users."""
permissions_by_key = _upsert_permissions(db)
roles_by_name = _upsert_roles(db, permissions_by_key)
_upsert_users(db, roles_by_name)