from django.contrib.auth import authenticate, login, logout from django.middleware.csrf import get_token from ninja import Router from ninja.errors import HttpError from apps.accounts.models import MenuItem from apps.accounts.permissions import PERMISSIONS from apps.accounts.schemas import CurrentUserOut, LoginIn router = Router(tags=["accounts"]) def permission_codes_for_user(user) -> list[str]: if user.is_superuser: return sorted(permission.codename for permission in PERMISSIONS) codes = set() for permission in user.get_all_permissions(): if permission.startswith("system."): codes.add(permission.removeprefix("system.")) return sorted(codes) def serialize_menu(menu: MenuItem, children: list[dict]) -> dict: return { "id": menu.id, "code": menu.code, "title": menu.title, "path": menu.path, "icon": menu.icon, "permission_code": menu.permission_code, "is_visible": menu.is_visible, "is_enabled": menu.is_enabled, "children": children, } def menu_tree_for_permissions(permission_codes: list[str]) -> list[dict]: allowed = set(permission_codes) menus = list( MenuItem.objects.filter(is_visible=True, is_enabled=True) .select_related("parent") .order_by("sort_order", "id") ) children_by_parent: dict[int | None, list[MenuItem]] = {} for menu in menus: if menu.permission_code and menu.permission_code not in allowed: continue children_by_parent.setdefault(menu.parent_id, []).append(menu) def build(parent_id: int | None) -> list[dict]: result = [] for menu in children_by_parent.get(parent_id, []): result.append(serialize_menu(menu, build(menu.id))) return result return build(None) def serialize_current_user(user) -> dict: permission_codes = permission_codes_for_user(user) return { "id": user.id, "username": user.username, "display_name": user.get_full_name() or user.username, "email": user.email, "is_superuser": user.is_superuser, "roles": [{"id": group.id, "name": group.name} for group in user.groups.all()], "user_groups": [ { "id": group.id, "name": group.name, "code": group.code, "description": group.description, "is_active": group.is_active, "created_at": group.created_at, "updated_at": group.updated_at, } for group in user.business_groups.all() ], "permissions": permission_codes, "menus": menu_tree_for_permissions(permission_codes), } def require_login(request): if not request.user.is_authenticated: raise HttpError(401, "Authentication required") return request.user def require_permission(request, codename: str): user = require_login(request) if user.is_superuser or user.has_perm(f"system.{codename}"): return user raise HttpError(403, "Permission denied") @router.get("/auth/csrf") def csrf(request): return {"csrfToken": get_token(request)} @router.post("/auth/login", response=CurrentUserOut) def login_user(request, payload: LoginIn): user = authenticate(request, username=payload.username, password=payload.password) if user is None or not user.is_active: raise HttpError(401, "Invalid username or password") login(request, user) return serialize_current_user(user) @router.post("/auth/logout") def logout_user(request): logout(request) return {"status": "ok"} @router.get("/auth/me", response=CurrentUserOut) def current_user(request): user = require_login(request) return serialize_current_user(user)