group_manage/ruoyi_common/base/serializer.py

# -*- coding: utf-8 -*-
# @Author  : YY

import dataclasses
from datetime import date
import decimal
import uuid
import typing as t
from flask import Response
from flask.json.provider import DefaultJSONProvider
from werkzeug.exceptions import HTTPException, default_exceptions
from werkzeug.http import http_date

from ruoyi_common.base.model import AjaxResponse

WSGIEnvironment: t.TypeAlias = dict[str, t.Any]


def _update_exceptions():
    """
    更新异常
    """
    for code in default_exceptions.keys():
        exc = default_exceptions[code]
        if isinstance(exc, HTTPException):
            new_exc = HttpException.from_http_exception(exc)
            default_exceptions[code] = new_exc
        else:
            continue


_update_exceptions()
del _update_exceptions


class HttpException(HTTPException):

    code: int | None = None
    description: str | None = None

    def __init__(
        self,
        description: str | None = None,
        response: Response | None = None,
    ) -> None:
        super().__init__()
        if description is not None:
            self.description = description
        self.response = response

    @classmethod
    def from_http_exception(cls, exc: HTTPException) -> "HttpException": 
        """
        从HTTPException转换为HttpException

        Args:
            exc (HTTPException): werkezeug的HTTPException

        Returns:
            HttpException: HttpException
        """
        error = cls(description=exc.description, response=exc.response)
        error.code = exc.code
        return error
    
    @property
    def name(self) -> str:
        """
        状态名称
        
        Returns:
            str: 状态名称
        """
        from werkzeug.http import HTTP_STATUS_CODES

        return HTTP_STATUS_CODES.get(self.code, "Unknown Error")  # type: ignore

    def get_description(
        self,
        environ: WSGIEnvironment | None = None,
        scope: dict[str, t.Any] | None = None,
    ) -> str:
        """
        异常描述
        
        Args:
            environ (WSGIEnvironment, optional): 环境变量. Defaults to None.
            scope (dict[str, t.Any], optional): 作用域. Defaults to None.

        Returns:
            str: 异常描述
        """
        return self.description or ""

    def get_body(
        self,
        environ: WSGIEnvironment | None = None,
        scope: dict[str, t.Any] | None = None,
    ) -> str:
        """
        异常响应体
        
        Args:
            environ (WSGIEnvironment, optional): 环境变量. Defaults to None.
            scope (dict[str, t.Any], optional): 作用域. Defaults to None.

        Returns:
            str: 异常响应体
        """
        ajax_resposne = AjaxResponse.from_error(msg=self.description)
        ajax_resposne.code = self.code
        return ajax_resposne.model_dump_json(
            exclude_unset = True,
            exclude_none = True,
        )

    def get_headers(
        self,
        environ: WSGIEnvironment | None = None,
        scope: dict[str, t.Any] | None = None,
    ) -> list[tuple[str, str]]:
        """
        异常请求头
        
        Args:
            environ (WSGIEnvironment, optional): 环境变量. Defaults to None.
            scope (dict[str, t.Any], optional): 作用域. Defaults to None.

        Returns:
            list[tuple[str, str]]: 异常请求头
        """
        return [("Content-Type", "application/json")]


def json_default(obj):
    """
    转化成可序列化对象

    Args:
        obj : 待序列化对象

    Returns:
        _type_: 可序列化对象
    """
    if isinstance(obj, date):
        return http_date(obj) 

    if isinstance(obj, decimal.Decimal):
        return str(obj)
    
    if isinstance(obj, uuid.UUID):
        return obj.hex

    if dataclasses and dataclasses.is_dataclass(obj):
        return dataclasses.asdict(obj)

    if isinstance(obj, AjaxResponse):
        return obj.model_dump_json()

    if hasattr(obj, "__html__"):
        return str(obj.__html__())

    raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")


class JsonProvider(DefaultJSONProvider):
    """
    自定义json序列化

    Args:
        DefaultJSONProvider: 默认flask的json序列化
    """
    
    default = staticmethod(json_default)


def handle_http_exception(error:HTTPException) -> Response:
    """
    处理http异常

    Args:
        error (HttpException): http异常

    Returns:
        ResponseReturnValue: 响应体
    """
    if not isinstance(error, HttpException):
        error = HttpException.from_http_exception(error)
    return error.get_response()