Files
mostovik-backend/src/apps/core/response.py
Aleksandr Meshchriakov f121445313 feat(core): add core module with mixins, services, and background jobs
- Add Model Mixins: TimestampMixin, SoftDeleteMixin, AuditMixin, etc.
- Add Base Services: BaseService, BulkOperationsMixin, QueryOptimizerMixin
- Add Base ViewSets with bulk operations
- Add BackgroundJob model for Celery task tracking
- Add BaseAppCommand for management commands
- Add permissions, pagination, filters, cache, logging
- Migrate tests to factory_boy + faker
- Add CHANGELOG.md
- 297 tests passing
2026-01-21 11:47:26 +01:00

161 lines
3.8 KiB
Python

"""
Unified API response wrapper.
Provides consistent response format across all API endpoints:
{
"success": bool,
"data": {...} | [...] | null,
"errors": [...] | null,
"meta": {
"request_id": "uuid",
"pagination": {...} // optional
}
}
"""
from typing import Any
from rest_framework import status
from rest_framework.response import Response
def api_response(
data: Any = None,
*,
status_code: int = status.HTTP_200_OK,
request_id: str | None = None,
pagination: dict[str, Any] | None = None,
headers: dict[str, str] | None = None,
) -> Response:
"""
Create a successful API response.
Args:
data: Response data (dict, list, or None)
status_code: HTTP status code (default 200)
request_id: Request tracking ID
pagination: Pagination metadata
headers: Additional response headers
Returns:
DRF Response with unified format
"""
meta = {}
if request_id:
meta["request_id"] = request_id
if pagination:
meta["pagination"] = pagination
response_data = {
"success": True,
"data": data,
"errors": None,
"meta": meta if meta else None,
}
return Response(response_data, status=status_code, headers=headers)
def api_error_response(
errors: list[dict[str, Any]],
*,
status_code: int = status.HTTP_400_BAD_REQUEST,
request_id: str | None = None,
headers: dict[str, str] | None = None,
) -> Response:
"""
Create an error API response.
Args:
errors: List of error dictionaries, each with 'code' and 'message'
status_code: HTTP status code (default 400)
request_id: Request tracking ID
headers: Additional response headers
Returns:
DRF Response with unified error format
"""
meta = {}
if request_id:
meta["request_id"] = request_id
response_data = {
"success": False,
"data": None,
"errors": errors,
"meta": meta if meta else None,
}
return Response(response_data, status=status_code, headers=headers)
def api_created_response(
data: Any = None,
*,
request_id: str | None = None,
headers: dict[str, str] | None = None,
) -> Response:
"""Shortcut for 201 Created response."""
return api_response(
data,
status_code=status.HTTP_201_CREATED,
request_id=request_id,
headers=headers,
)
def api_no_content_response(
*,
request_id: str | None = None,
headers: dict[str, str] | None = None,
) -> Response:
"""Shortcut for 204 No Content response."""
meta = {}
if request_id:
meta["request_id"] = request_id
return Response(
{"success": True, "data": None, "errors": None, "meta": meta if meta else None},
status=status.HTTP_204_NO_CONTENT,
headers=headers,
)
def api_paginated_response(
data: list[Any],
*,
page: int,
page_size: int,
total_count: int,
request_id: str | None = None,
headers: dict[str, str] | None = None,
) -> Response:
"""
Create a paginated API response.
Args:
data: List of items for current page
page: Current page number
page_size: Number of items per page
total_count: Total number of items
request_id: Request tracking ID
headers: Additional response headers
"""
total_pages = (total_count + page_size - 1) // page_size if page_size > 0 else 0
pagination = {
"page": page,
"page_size": page_size,
"total_count": total_count,
"total_pages": total_pages,
"has_next": page < total_pages,
"has_previous": page > 1,
}
return api_response(
data,
request_id=request_id,
pagination=pagination,
headers=headers,
)