280 lines
11 KiB
Python
280 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, is_dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Callable, Dict, Optional, Tuple, Union
|
|
import uuid
|
|
|
|
from flask import jsonify, make_response, request, Flask, Response
|
|
|
|
|
|
class ApiResponder:
|
|
"""
|
|
Centralized JSON response builder for Flask APIs.
|
|
|
|
This class enforces a consistent envelope for all responses:
|
|
|
|
{
|
|
"app": "<APP NAME>",
|
|
"version": "<APP VERSION>",
|
|
"status": <HTTP STATUS CODE>,
|
|
"timestamp": "<UTC ISO8601>",
|
|
"request_id": "<optional request id>",
|
|
"result": <your data OR null>,
|
|
"error": {
|
|
"code": "<optional machine code>",
|
|
"message": "<human message>",
|
|
"details": {...} # optional extras (validation fields, etc.)
|
|
},
|
|
"meta": { ... } # optional metadata (pagination, etc.)
|
|
}
|
|
|
|
Usage:
|
|
responder = ApiResponder(app_name="Code of Conquest",
|
|
version_provider=lambda: CURRENT_VERSION)
|
|
|
|
return responder.ok({"hello": "world"})
|
|
return responder.created({"id": 123})
|
|
return responder.bad_request("Missing field `name`", details={"field": "name"})
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
app_name: str,
|
|
version_provider: str,
|
|
include_request_id: bool = True,
|
|
default_headers: Optional[Dict[str, str]] = None,
|
|
) -> None:
|
|
"""
|
|
:param app_name: Human-friendly app name included in every response.
|
|
:param version_provider: Callable returning a version string at call time.
|
|
:param include_request_id: When True, include request id (from X-Request-ID header if present, else generated).
|
|
:param default_headers: Extra headers to attach to every response (e.g., CORS, caching).
|
|
"""
|
|
self.app_name = app_name
|
|
self.version_provider = version_provider
|
|
self.include_request_id = include_request_id
|
|
self.default_headers = default_headers or {}
|
|
|
|
# ---------- Public helpers for common statuses ----------
|
|
|
|
def ok(self, result: Any = None, meta: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""200 OK."""
|
|
return self._build(status=200, result=result, meta=meta, headers=headers)
|
|
|
|
def created(self, result: Any = None, meta: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""201 Created."""
|
|
return self._build(status=201, result=result, meta=meta, headers=headers)
|
|
|
|
def accepted(self, result: Any = None, meta: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""202 Accepted."""
|
|
return self._build(status=202, result=result, meta=meta, headers=headers)
|
|
|
|
def no_content(self, headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""
|
|
204 No Content. Returns the standard envelope with result=null for consistency.
|
|
(If you prefer an empty body, switch to make_response(("", 204)) in your code.)
|
|
"""
|
|
return self._build(status=204, result=None, headers=headers)
|
|
|
|
def bad_request(self, message: str, code: Optional[str] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""400 Bad Request."""
|
|
return self._build_error(400, message, code, details, headers)
|
|
|
|
def unauthorized(self, message: str = "Unauthorized", code: Optional[str] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""401 Unauthorized."""
|
|
return self._build_error(401, message, code, details, headers)
|
|
|
|
def forbidden(self, message: str = "Forbidden", code: Optional[str] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""403 Forbidden."""
|
|
return self._build_error(403, message, code, details, headers)
|
|
|
|
def not_found(self, message: str = "Not Found", code: Optional[str] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""404 Not Found."""
|
|
return self._build_error(404, message, code, details, headers)
|
|
|
|
def conflict(self, message: str = "Conflict", code: Optional[str] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""409 Conflict."""
|
|
return self._build_error(409, message, code, details, headers)
|
|
|
|
def unprocessable(self, message: str = "Unprocessable Entity", code: Optional[str] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""422 Unprocessable Entity (great for validation errors)."""
|
|
return self._build_error(422, message, code, details, headers)
|
|
|
|
def error(self, message: str = "Internal Server Error", code: Optional[str] = None,
|
|
details: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""500 Internal Server Error."""
|
|
return self._build_error(500, message, code, details, headers)
|
|
|
|
# ---------- Pagination helper ----------
|
|
|
|
def paginate(self,
|
|
items: Any,
|
|
total: int,
|
|
page: int,
|
|
per_page: int,
|
|
extra_meta: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""
|
|
200 OK with pagination metadata.
|
|
|
|
:param items: The current page of items (list or serializable value).
|
|
:param total: Total count of items across all pages.
|
|
:param page: 1-based page index.
|
|
:param per_page: Page size.
|
|
:param extra_meta: Optional extra metadata to merge into the meta block.
|
|
"""
|
|
# Build pagination metadata explicitly for clarity (no list comps).
|
|
meta: Dict[str, Any] = {}
|
|
meta["total"] = int(total)
|
|
meta["page"] = int(page)
|
|
meta["per_page"] = int(per_page)
|
|
|
|
# Compute total pages carefully with integer math.
|
|
total_pages = int((total + per_page - 1) // per_page) if per_page > 0 else 0
|
|
meta["total_pages"] = total_pages
|
|
|
|
if extra_meta is not None:
|
|
for key in extra_meta:
|
|
meta[key] = extra_meta[key]
|
|
|
|
return self._build(status=200, result=items, meta=meta, headers=headers)
|
|
|
|
# ---------- Exception binding (optional but handy) ----------
|
|
|
|
def register_error_handlers(self, app: Flask) -> None:
|
|
"""
|
|
Registers generic error handlers that convert exceptions into standard JSON.
|
|
Override selectively in your app as needed.
|
|
"""
|
|
|
|
@app.errorhandler(400)
|
|
def _h400(e): # pragma: no cover
|
|
return self.bad_request(getattr(e, "description", "Bad Request"))
|
|
|
|
@app.errorhandler(401)
|
|
def _h401(e): # pragma: no cover
|
|
return self.unauthorized(getattr(e, "description", "Unauthorized"))
|
|
|
|
@app.errorhandler(403)
|
|
def _h403(e): # pragma: no cover
|
|
return self.forbidden(getattr(e, "description", "Forbidden"))
|
|
|
|
@app.errorhandler(404)
|
|
def _h404(e): # pragma: no cover
|
|
return self.not_found(getattr(e, "description", "Not Found"))
|
|
|
|
@app.errorhandler(422)
|
|
def _h422(e): # pragma: no cover
|
|
message = getattr(e, "description", "Unprocessable Entity")
|
|
# Marshmallow/WTF often attach data on e.data; include if present.
|
|
details = getattr(e, "data", None)
|
|
return self.unprocessable(message=message, details=details)
|
|
|
|
@app.errorhandler(500)
|
|
def _h500(e): # pragma: no cover
|
|
return self.error()
|
|
|
|
# ---------- Core builder ----------
|
|
|
|
def _build(self,
|
|
status: int,
|
|
result: Any = None,
|
|
meta: Optional[Dict[str, Any]] = None,
|
|
headers: Optional[Dict[str, str]] = None) -> Tuple[Response, int]:
|
|
"""
|
|
Build the canonical JSON body and Response object.
|
|
"""
|
|
# Convert dataclasses to plain dicts to keep jsonify happy.
|
|
safe_result = self._to_plain(result)
|
|
|
|
body: Dict[str, Any] = {}
|
|
body["app"] = self.app_name
|
|
body["version"] = self.version_provider or ""
|
|
body["status"] = int(status)
|
|
body["timestamp"] = datetime.now(timezone.utc).isoformat()
|
|
|
|
if self.include_request_id:
|
|
# Prefer inbound request id if the client provided one.
|
|
req_id = request.headers.get("X-Request-ID")
|
|
if req_id is None or req_id == "":
|
|
req_id = str(uuid.uuid4())
|
|
body["request_id"] = req_id
|
|
|
|
body["result"] = safe_result
|
|
|
|
if meta is not None:
|
|
body["meta"] = meta
|
|
|
|
response = make_response(jsonify(body), status)
|
|
|
|
# Attach default headers first, then per-call overrides.
|
|
for key in self.default_headers:
|
|
response.headers[key] = self.default_headers[key]
|
|
if headers is not None:
|
|
for key in headers:
|
|
response.headers[key] = headers[key]
|
|
|
|
return response, status
|
|
|
|
def _build_error(self,
|
|
status: int,
|
|
message: str,
|
|
code: Optional[str],
|
|
details: Optional[Dict[str, Any]],
|
|
headers: Optional[Dict[str, str]]) -> Tuple[Response, int]:
|
|
"""
|
|
Build a standardized error envelope.
|
|
"""
|
|
error_block: Dict[str, Any] = {}
|
|
error_block["message"] = message
|
|
|
|
if code is not None and code != "":
|
|
error_block["code"] = code
|
|
|
|
if details is not None:
|
|
# Convert nested dataclasses if any.
|
|
error_block["details"] = self._to_plain(details)
|
|
|
|
# Errors carry result=null
|
|
return self._build(status=status, result=None, meta={"error": error_block}, headers=headers)
|
|
|
|
def _to_plain(self, value: Any) -> Any:
|
|
"""
|
|
Convert dataclasses to dicts recursively; leave other JSON-serializable values as-is.
|
|
"""
|
|
if is_dataclass(value):
|
|
return asdict(value)
|
|
|
|
# Handle lists/tuples without comprehensions for clarity.
|
|
if isinstance(value, (list, tuple)):
|
|
converted = []
|
|
for item in value:
|
|
converted.append(self._to_plain(item))
|
|
return converted
|
|
|
|
if isinstance(value, dict):
|
|
converted_dict: Dict[str, Any] = {}
|
|
for k in value:
|
|
converted_dict[k] = self._to_plain(value[k])
|
|
return converted_dict
|
|
|
|
# Let Flask jsonify handle the rest (numbers, strings, None, bool).
|
|
return value
|