123 lines
4.6 KiB
Python
123 lines
4.6 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 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
|
|
|
|
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_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"),
|
|
("audit_log:read", "View audit log"),
|
|
]
|
|
|
|
# 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_types:read", "sub_part_types:read", "imported_parts: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
|
|
|
|
# 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())
|