179 lines
6.9 KiB
Python
179 lines
6.9 KiB
Python
"""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", "Ocular", "Rail"]
|
|
|
|
PART_TYPES = ["Flip Cover", "ARD", "Polarizer", "Adapter Ring", "Scope"]
|
|
|
|
SUB_PART_TYPES = {
|
|
"Scope": ["Riflescope", "Binoculars", "Combat Optic", "Red Dot", "Spotter"],
|
|
}
|
|
|
|
# Default permission catalogue: "<resource>:<action>".
|
|
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())
|