CODE_ASSISTANT/5_django_backend/common/envelope.py

"""
응답 envelope — 프론트 `types/api.ts` 의 CommonResponse 와 1:1.

{
  success, statusCode, code, message, data, counts, errors, timestamp, meta
}

- DRF 뷰: 그냥 `Response(data)` 하면 EnvelopeRenderer 가 감쌈.
  목록은 `Response(paged(items, page, limit, total))` — data=items, meta 채워짐.
- DRF 예외: envelope_exception_handler 가 success:false 로 감쌈. code 는 CodedError 로 지정.
- 순수 Django async 뷰(스트림): `json_error()` 로 직접 만듦.
"""

from datetime import datetime, timezone

from django.http import JsonResponse
from rest_framework.exceptions import APIException
from rest_framework.renderers import JSONRenderer

_LIST_MARK = "__paged__"


def _now() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


def envelope(
    *,
    success: bool,
    status_code: int,
    data=None,
    code: str | None = None,
    message: str | None = None,
    errors: list[str] | None = None,
    counts: int | None = None,
    meta: dict | None = None,
) -> dict:
    return {
        "success": success,
        "statusCode": status_code,
        "code": code,
        "message": message,
        "data": data,
        "counts": counts,
        "errors": errors or [],
        "timestamp": _now(),
        "meta": meta,
    }


def paged(items: list, page: int, limit: int, total: int) -> dict:
    """목록 응답용. Renderer 가 data=items, meta=페이지 정보로 풀어줌."""
    pages = (total + limit - 1) // limit if limit > 0 else 0
    return {
        _LIST_MARK: True,
        "items": items,
        "counts": total,
        "meta": {
            "currentPage": page,
            "pageSize": limit,
            "totalItems": total,
            "totalPages": pages,
            "hasNextPage": page < pages,
            "hasPreviousPage": page > 1,
        },
    }


def _is_wrapped(data) -> bool:
    return isinstance(data, dict) and "success" in data and "statusCode" in data


class EnvelopeRenderer(JSONRenderer):
    def render(self, data, accepted_media_type=None, renderer_context=None):
        response = (renderer_context or {}).get("response")
        status_code = getattr(response, "status_code", 200)
        if _is_wrapped(data):
            body = data
        elif isinstance(data, dict) and data.get(_LIST_MARK):
            body = envelope(
                success=True,
                status_code=status_code,
                data=data["items"],
                counts=data["counts"],
                meta=data["meta"],
            )
        else:
            body = envelope(success=True, status_code=status_code, data=data)
        return super().render(body, accepted_media_type, renderer_context)


class CodedError(APIException):
    """상태코드 + 프론트가 분기하는 code 문자열을 같이 던짐."""

    def __init__(self, status_code: int, code: str, message: str):
        super().__init__(detail=message)
        self.status_code = status_code
        self.code_name = code


_DEFAULT_CODES = {
    400: "VALIDATION_ERROR",
    401: "UNAUTHORIZED",
    403: "FORBIDDEN",
    404: "NOT_FOUND",
    405: "METHOD_NOT_ALLOWED",
    409: "CONFLICT",
    501: "NOT_IMPLEMENTED",
    502: "UPSTREAM_ERROR",
    503: "UPSTREAM_UNAVAILABLE",
}


def _flatten(data) -> str:
    """serializer 검증 오류(dict/list)를 한 줄로."""
    if isinstance(data, str):
        return data
    if isinstance(data, list):
        return " ".join(_flatten(d) for d in data)
    if isinstance(data, dict):
        parts = []
        for k, v in data.items():
            msg = _flatten(v)
            parts.append(msg if k in ("detail", "non_field_errors") else f"{k}: {msg}")
        return " ".join(parts)
    return str(data)


def envelope_exception_handler(exc, context):
    # rest_framework.views 는 settings 의 renderer(=이 모듈)를 import 하므로 여기서 늦게 import (순환 방지)
    from rest_framework.views import exception_handler

    response = exception_handler(exc, context)
    if response is None:
        return None
    code = getattr(exc, "code_name", None) or _DEFAULT_CODES.get(response.status_code, "ERROR")
    message = _flatten(response.data)
    response.data = envelope(
        success=False,
        status_code=response.status_code,
        code=code,
        message=message,
        errors=[message],
    )
    return response


def json_error(status_code: int, code: str, message: str) -> JsonResponse:
    """DRF 안 거치는 async 뷰용."""
    return JsonResponse(
        envelope(success=False, status_code=status_code, code=code, message=message, errors=[message]),
        status=status_code,
        json_dumps_params={"ensure_ascii": False},
    )