feat: Add analytics, incident, and alert models and APIs, along with user permissions and bulk profile lookups.

This commit is contained in:
pacnpal
2026-01-07 11:07:36 -05:00
parent 4da7e52fb0
commit 3ec5a4857d
46 changed files with 4012 additions and 199 deletions

View File

@@ -826,6 +826,63 @@ def check_user_deletion_eligibility(request, user_id):
# === USER PROFILE ENDPOINTS ===
@extend_schema(
operation_id="get_user_permissions",
summary="Get current user's management permissions",
description="Get the authenticated user's management permissions including role information.",
responses={
200: {
"description": "User permissions",
"example": {
"user_id": "uuid",
"is_superuser": True,
"is_staff": True,
"is_moderator": False,
"roles": ["admin"],
"permissions": ["can_moderate", "can_manage_users"],
},
},
401: {
"description": "Authentication required",
"example": {"detail": "Authentication credentials were not provided."},
},
},
tags=["User Profile"],
)
@api_view(["GET"])
@permission_classes([IsAuthenticated])
def get_user_permissions(request):
"""Get the authenticated user's management permissions."""
user = request.user
profile = getattr(user, "profile", None)
# Get roles from profile if exists
roles = []
if profile:
if hasattr(profile, "role") and profile.role:
roles.append(profile.role)
if user.is_superuser:
roles.append("admin")
if user.is_staff:
roles.append("staff")
# Build permissions list based on flags
permissions = []
if user.is_superuser or user.is_staff:
permissions.extend(["can_moderate", "can_manage_users", "can_view_admin"])
elif profile and getattr(profile, "is_moderator", False):
permissions.append("can_moderate")
return Response({
"user_id": str(user.id),
"is_superuser": user.is_superuser,
"is_staff": user.is_staff,
"is_moderator": profile and getattr(profile, "is_moderator", False) if profile else False,
"roles": list(set(roles)), # Deduplicate
"permissions": list(set(permissions)), # Deduplicate
}, status=status.HTTP_200_OK)
@extend_schema(
operation_id="get_user_profile",
summary="Get current user's complete profile",
@@ -935,8 +992,8 @@ def get_user_preferences(request):
"allow_messages": user.allow_messages,
}
serializer = UserPreferencesSerializer(data=data)
return Response(serializer.data, status=status.HTTP_200_OK)
# Return the data directly - no validation needed for GET response
return Response(data, status=status.HTTP_200_OK)
@extend_schema(
@@ -1056,8 +1113,8 @@ def get_notification_settings(request):
},
}
serializer = NotificationSettingsSerializer(data=data)
return Response(serializer.data, status=status.HTTP_200_OK)
# Return the data directly - no validation needed for GET response
return Response(data, status=status.HTTP_200_OK)
@extend_schema(
@@ -1131,8 +1188,8 @@ def get_privacy_settings(request):
"allow_messages": user.allow_messages,
}
serializer = PrivacySettingsSerializer(data=data)
return Response(serializer.data, status=status.HTTP_200_OK)
# Return the data directly - no validation needed for GET response
return Response(data, status=status.HTTP_200_OK)
@extend_schema(
@@ -1198,8 +1255,8 @@ def get_security_settings(request):
"active_sessions": getattr(user, "active_sessions", 1),
}
serializer = SecuritySettingsSerializer(data=data)
return Response(serializer.data, status=status.HTTP_200_OK)
# Return the data directly - no validation needed for GET response
return Response(data, status=status.HTTP_200_OK)
@extend_schema(
@@ -1273,8 +1330,8 @@ def get_user_statistics(request):
"last_activity": user.last_login,
}
serializer = UserStatisticsSerializer(data=data)
return Response(serializer.data, status=status.HTTP_200_OK)
# Return the data directly - no validation needed for GET response
return Response(data, status=status.HTTP_200_OK)
# === TOP LISTS ENDPOINTS ===
@@ -1732,3 +1789,135 @@ def cancel_email_change(request):
},
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
@extend_schema(
operation_id="bulk_get_profiles",
summary="Get multiple user profiles by user IDs",
description="Fetch profile information for multiple users at once. Useful for displaying user info in lists.",
parameters=[
OpenApiParameter(
name="user_ids",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Comma-separated list of user IDs",
required=True,
),
],
responses={
200: {
"description": "List of user profiles",
"example": [
{
"user_id": "123",
"username": "john_doe",
"display_name": "John Doe",
"avatar_url": "https://example.com/avatar.jpg",
}
],
},
},
tags=["User Profile"],
)
@api_view(["GET"])
@permission_classes([IsAuthenticated])
def bulk_get_profiles(request):
"""Get multiple user profiles by IDs for efficient bulk lookups."""
user_ids_param = request.query_params.get("user_ids", "")
if not user_ids_param:
return Response([], status=status.HTTP_200_OK)
user_ids = [uid.strip() for uid in user_ids_param.split(",") if uid.strip()]
if not user_ids:
return Response([], status=status.HTTP_200_OK)
# Limit to prevent abuse
if len(user_ids) > 100:
user_ids = user_ids[:100]
profiles = UserProfile.objects.filter(user__user_id__in=user_ids).select_related("user", "avatar")
result = []
for profile in profiles:
result.append({
"user_id": str(profile.user.user_id),
"username": profile.user.username,
"display_name": profile.display_name,
"avatar_url": profile.get_avatar_url() if hasattr(profile, "get_avatar_url") else None,
})
return Response(result, status=status.HTTP_200_OK)
@extend_schema(
operation_id="get_users_with_emails",
summary="Get users with email addresses (admin/moderator only)",
description="Fetch user information including emails. Restricted to admins and moderators.",
parameters=[
OpenApiParameter(
name="user_ids",
type=OpenApiTypes.STR,
location=OpenApiParameter.QUERY,
description="Comma-separated list of user IDs",
required=True,
),
],
responses={
200: {
"description": "List of users with emails",
"example": [
{
"user_id": "123",
"username": "john_doe",
"email": "john@example.com",
"display_name": "John Doe",
}
],
},
403: {"description": "Not authorized - admin or moderator access required"},
},
tags=["User Management"],
)
@api_view(["GET"])
@permission_classes([IsAuthenticated])
def get_users_with_emails(request):
"""Get users with email addresses - restricted to admins and moderators."""
user = request.user
# Check if user is admin or moderator
if not (user.is_staff or user.is_superuser or getattr(user, "role", "") in ["ADMIN", "MODERATOR"]):
return Response(
{"detail": "Admin or moderator access required"},
status=status.HTTP_403_FORBIDDEN,
)
user_ids_param = request.query_params.get("user_ids", "")
if not user_ids_param:
return Response([], status=status.HTTP_200_OK)
user_ids = [uid.strip() for uid in user_ids_param.split(",") if uid.strip()]
if not user_ids:
return Response([], status=status.HTTP_200_OK)
# Limit to prevent abuse
if len(user_ids) > 100:
user_ids = user_ids[:100]
users = User.objects.filter(user_id__in=user_ids).select_related("profile")
result = []
for u in users:
profile = getattr(u, "profile", None)
result.append({
"user_id": str(u.user_id),
"username": u.username,
"email": u.email,
"display_name": profile.display_name if profile else None,
})
return Response(result, status=status.HTTP_200_OK)