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": "", "version": "", "status": , "timestamp": "", "request_id": "", "result": , "error": { "code": "", "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