from django.db.models import Q from ninja import Query, Router, Schema from apps.users.models import Major, University from core.api.schemas import ErrorSchema, MessageSchema from core.authentication import jwt_auth meta_router = Router(tags=['meta']) class MetaOptionSchema(Schema): id: int code: str label: str is_active: bool = True user_count: int = 0 class PagedMetaOptionSchema(Schema): count: int results: list[MetaOptionSchema] class MetaOptionWriteSchema(Schema): code: str name: str is_active: bool = True def _is_staff(user): return bool(user and (user.is_staff or user.is_superuser)) def _option_payload(obj, user_count=0): return { "id": obj.id, "code": obj.code, "label": obj.name, "is_active": obj.is_active, "user_count": user_count, } def _list_options(model, search, limit, offset, active_only=True): queryset = model.objects.filter(is_deleted=False).order_by("name") if active_only: queryset = queryset.filter(is_active=True) if search: queryset = queryset.filter(Q(code__icontains=search) | Q(name__icontains=search)) count = queryset.count() return count, list(queryset[offset : offset + limit]) @meta_router.get("/majors", response=PagedMetaOptionSchema) def list_majors( request, search: str | None = Query(None), limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), ): count, majors = _list_options(Major, search, limit, offset) return {"count": count, "results": [_option_payload(m) for m in majors]} @meta_router.get("/universities", response=PagedMetaOptionSchema) def list_universities( request, search: str | None = Query(None), limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), ): count, universities = _list_options(University, search, limit, offset) return {"count": count, "results": [_option_payload(u) for u in universities]} @meta_router.get("/admin/majors", response={200: PagedMetaOptionSchema, 403: ErrorSchema}, auth=jwt_auth) def admin_list_majors( request, search: str | None = Query(None), limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), ): if not _is_staff(request.auth): return 403, {"error": "Permission denied"} count, majors = _list_options(Major, search, limit, offset, active_only=False) return 200, { "count": count, "results": [_option_payload(m, m.users.filter(is_deleted=False).count()) for m in majors], } @meta_router.post("/admin/majors", response={201: MetaOptionSchema, 403: ErrorSchema, 400: ErrorSchema}, auth=jwt_auth) def admin_create_major(request, payload: MetaOptionWriteSchema): if not _is_staff(request.auth): return 403, {"error": "Permission denied"} if Major.all_objects.filter(code=payload.code).exists(): return 400, {"error": "Major code already exists"} major = Major.objects.create(code=payload.code.strip(), name=payload.name.strip(), is_active=payload.is_active) return 201, _option_payload(major) @meta_router.put("/admin/majors/{int:item_id}", response={200: MetaOptionSchema, 403: ErrorSchema, 400: ErrorSchema}, auth=jwt_auth) def admin_update_major(request, item_id: int, payload: MetaOptionWriteSchema): if not _is_staff(request.auth): return 403, {"error": "Permission denied"} try: major = Major.objects.get(id=item_id) except Major.DoesNotExist: return 400, {"error": "Major not found"} conflict = Major.all_objects.filter(code=payload.code).exclude(id=item_id).exists() if conflict: return 400, {"error": "Major code already exists"} major.code = payload.code.strip() major.name = payload.name.strip() major.is_active = payload.is_active major.save(update_fields=["code", "name", "is_active", "updated_at"]) return 200, _option_payload(major, major.users.filter(is_deleted=False).count()) @meta_router.delete("/admin/majors/{int:item_id}", response={200: MessageSchema, 403: ErrorSchema, 400: ErrorSchema}, auth=jwt_auth) def admin_delete_major(request, item_id: int): if not request.auth.is_superuser: return 403, {"error": "Only superusers can delete majors"} try: major = Major.objects.get(id=item_id) except Major.DoesNotExist: return 400, {"error": "Major not found"} major.delete() return 200, {"message": "Major deleted"} @meta_router.post("/admin/majors/{int:item_id}/restore", response={200: MessageSchema, 403: ErrorSchema, 400: ErrorSchema}, auth=jwt_auth) def admin_restore_major(request, item_id: int): if not request.auth.is_superuser: return 403, {"error": "Only superusers can restore majors"} try: major = Major.deleted_objects.get(id=item_id) except Major.DoesNotExist: return 400, {"error": "Major not found"} major.restore() return 200, {"message": "Major restored"} @meta_router.get("/admin/universities", response={200: PagedMetaOptionSchema, 403: ErrorSchema}, auth=jwt_auth) def admin_list_universities( request, search: str | None = Query(None), limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), ): if not _is_staff(request.auth): return 403, {"error": "Permission denied"} count, universities = _list_options(University, search, limit, offset, active_only=False) return 200, { "count": count, "results": [_option_payload(u, u.users.filter(is_deleted=False).count()) for u in universities], } @meta_router.post("/admin/universities", response={201: MetaOptionSchema, 403: ErrorSchema, 400: ErrorSchema}, auth=jwt_auth) def admin_create_university(request, payload: MetaOptionWriteSchema): if not _is_staff(request.auth): return 403, {"error": "Permission denied"} if University.all_objects.filter(code=payload.code).exists(): return 400, {"error": "University code already exists"} university = University.objects.create(code=payload.code.strip(), name=payload.name.strip(), is_active=payload.is_active) return 201, _option_payload(university) @meta_router.put("/admin/universities/{int:item_id}", response={200: MetaOptionSchema, 403: ErrorSchema, 400: ErrorSchema}, auth=jwt_auth) def admin_update_university(request, item_id: int, payload: MetaOptionWriteSchema): if not _is_staff(request.auth): return 403, {"error": "Permission denied"} try: university = University.objects.get(id=item_id) except University.DoesNotExist: return 400, {"error": "University not found"} conflict = University.all_objects.filter(code=payload.code).exclude(id=item_id).exists() if conflict: return 400, {"error": "University code already exists"} university.code = payload.code.strip() university.name = payload.name.strip() university.is_active = payload.is_active university.save(update_fields=["code", "name", "is_active", "updated_at"]) return 200, _option_payload(university, university.users.filter(is_deleted=False).count()) @meta_router.delete("/admin/universities/{int:item_id}", response={200: MessageSchema, 403: ErrorSchema, 400: ErrorSchema}, auth=jwt_auth) def admin_delete_university(request, item_id: int): if not request.auth.is_superuser: return 403, {"error": "Only superusers can delete universities"} try: university = University.objects.get(id=item_id) except University.DoesNotExist: return 400, {"error": "University not found"} university.delete() return 200, {"message": "University deleted"} @meta_router.post("/admin/universities/{int:item_id}/restore", response={200: MessageSchema, 403: ErrorSchema, 400: ErrorSchema}, auth=jwt_auth) def admin_restore_university(request, item_id: int): if not request.auth.is_superuser: return 403, {"error": "Only superusers can restore universities"} try: university = University.deleted_objects.get(id=item_id) except University.DoesNotExist: return 400, {"error": "University not found"} university.restore() return 200, {"message": "University restored"}