"""Bootstrap default permissions, roles, and the first superuser. Idempotent: safe to run repeatedly. Reads the superuser credentials from settings (env vars or .env). Usage: uv run python -m scripts.seed """ import asyncio from app.core.config import settings from app.core.database import AsyncSessionLocal, init_db from app.crud import attachment_point_name as apn_crud from app.crud import part_type as part_type_crud from app.crud import role as role_crud from app.crud import sub_part_type as sub_part_type_crud from app.crud import user as user_crud ATTACHMENT_POINT_NAMES = ["Objective", "Occular"] PART_TYPES = ["Flip Cover", "ARD", "Polarizer", "Adapter Ring", "Scope"] SUB_PART_TYPES = { "Scope": ["Riflescope", "Binoculars", "Combat Optic", "Red Dot", "Spotter"], } # Default permission catalogue: ":". PERMISSIONS = [ ("users:read", "View users"), ("users:write", "Create and update users"), ("users:delete", "Delete users"), ("roles:read", "View roles and permissions"), ("roles:write", "Create, update and delete roles"), ("iso_threads:read", "View ISO threads"), ("iso_threads:write", "Create and update ISO threads"), ("iso_threads:delete", "Delete ISO threads"), ("unc_threads:read", "View UNC threads"), ("unc_threads:write", "Create and update UNC threads"), ("unc_threads:delete", "Delete UNC threads"), ("manufacturers:read", "View manufacturers"), ("manufacturers:write", "Create and update manufacturers"), ("manufacturers:delete", "Delete manufacturers"), ("scope_models:read", "View scope models"), ("scope_models:write", "Create and update scope models"), ("scope_models:delete", "Delete scope models"), ("part_colours:read", "View part colours"), ("part_colours:write", "Create and update part colours"), ("part_colours:delete", "Delete part colours"), ("part_types:read", "View part types"), ("part_types:write", "Create and update part types"), ("part_types:delete", "Delete part types"), ("sub_part_types:read", "View sub part types"), ("sub_part_types:write", "Create and update sub part types"), ("sub_part_types:delete", "Delete sub part types"), ("imported_parts:read", "View imported parts"), ("imported_parts:write", "Sync and configure imported parts"), ("imported_parts:delete", "Delete imported parts"), ("static_connections:read", "View static connections"), ("static_connections:write", "Create and update static connections"), ("static_connections:delete", "Delete static connections"), ("clamp_connections:read", "View clamp connections"), ("clamp_connections:write", "Create and update clamp connections"), ("clamp_connections:delete", "Delete clamp connections"), ("attachment_point_names:read", "View attachment point names"), ("attachment_point_names:write", "Create and update attachment point names"), ("attachment_point_names:delete", "Delete attachment point names"), ("attachment_points:read", "View attachment points"), ("attachment_points:write", "Create and update attachment points"), ("attachment_points:delete", "Delete attachment points"), ("augmented_parts:read", "View augmented parts"), ("augmented_parts:write", "Create and update augmented parts"), ("augmented_parts:delete", "Delete augmented parts"), ("fits:read", "View part fits"), ("fits:write", "Create, promote and recompute part fits"), ("fits:delete", "Delete part fits"), ("audit_log:read", "View audit log"), ] # Role auto-assigned to new (registered/LDAP-provisioned) users unless an admin # marks a different one as default in the UI. DEFAULT_ROLE = "user" # Default roles -> permission names. "admin" gets everything. ROLES = { "admin": [name for name, _ in PERMISSIONS], # includes audit_log:read "user": [ "users:read", "iso_threads:read", "unc_threads:read", "manufacturers:read", "scope_models:read", "part_colours:read", "part_types:read", "sub_part_types:read", "imported_parts:read", "static_connections:read", "clamp_connections:read", "attachment_point_names:read", "attachment_points:read", "augmented_parts:read", "fits:read", ], } async def seed() -> None: await init_db() async with AsyncSessionLocal() as db: # Permissions perms = {} for name, description in PERMISSIONS: perm = await role_crud.get_permission_by_name(db, name) if perm is None: perm = await role_crud.create_permission( db, name=name, description=description ) perms[name] = perm # Roles roles = {} for role_name, perm_names in ROLES.items(): role = await role_crud.get_by_name(db, role_name) if role is None: role = await role_crud.create( db, name=role_name, description=f"{role_name.title()} role", permission_ids=[perms[p].id for p in perm_names], ) roles[role_name] = role # Ensure a default role is marked (idempotent). if await role_crud.get_default(db) is None: default_role = roles.get(DEFAULT_ROLE) if default_role is not None: await role_crud.set_default(db, default_role) # Attachment point names for apn_name in ATTACHMENT_POINT_NAMES: if await apn_crud.get_by_name(db, apn_name) is None: await apn_crud.create(db, name=apn_name) # Part types for pt_name in PART_TYPES: if await part_type_crud.get_by_name(db, pt_name) is None: await part_type_crud.create(db, name=pt_name) # Sub part types for pt_name, spt_names in SUB_PART_TYPES.items(): pt = await part_type_crud.get_by_name(db, pt_name) if pt is not None: for spt_name in spt_names: if ( await sub_part_type_crud.get_by_name_and_part_type( db, spt_name, pt.id ) is None ): await sub_part_type_crud.create( db, name=spt_name, part_type_id=pt.id ) # Superuser existing = await user_crud.get_by_email(db, settings.first_superuser_email) if existing is None: await user_crud.create( db, email=settings.first_superuser_email, password=settings.first_superuser_password, full_name="Administrator", is_superuser=True, role_ids=[roles["admin"].id], ) print(f"Created superuser: {settings.first_superuser_email}") else: print(f"Superuser already exists: {settings.first_superuser_email}") print("Seed complete.") if __name__ == "__main__": asyncio.run(seed())