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