Create fastapi_contrib.exception_handlers.py

fastapi_contrib.exception_handlers.py

@@ -0,0 +1,164 @@
+from typing import Any, List, Optional
+
+from fastapi import FastAPI
+from fastapi.exceptions import RequestValidationError
+from pydantic import EnumError, StrRegexError
+from starlette.exceptions import HTTPException as StarletteHTTPException
+from starlette.requests import Request
+
+from fastapi_contrib.common.responses import UJSONResponse
+
+def parse_error(
+    err: Any, field_names: List, raw: bool = True
+) -> Optional[dict]:
+
+    if isinstance(err.exc, EnumError):
+        permitted_values = ", ".join(
+            [f"'{val}'" for val in err.exc.enum_values]
+        )
+        message = f"Value is not a valid enumeration member; " \
+                  f"permitted: {permitted_values}."
+    elif isinstance(err.exc, StrRegexError):
+        message = "Provided value doesn't match valid format."
+    else:
+        message = str(err.exc) or ""
+
+    if hasattr(err.exc, "code") and err.exc.code.startswith("error_code"):
+        error_code = int(err.exc.code.split(".")[-1])
+    else:
+        # default error code for non-custom errors is 400
+        error_code = 400
+
+    if not raw:
+        if len(err.loc_tuple()) == 2:
+            if str(err.loc_tuple()[0]) in ["body", "query"]:
+                name = err.loc_tuple()[1]
+            else:
+                name = err.loc_tuple()[0]
+        elif len(err.loc_tuple()) == 1:
+            if str(err.loc_tuple()[0]) == "body":
+                name = "__all__"
+            else:
+                name = str(err.loc_tuple()[0])
+        else:   
+            name = "__all__"
+    else:
+        if len(err.loc_tuple()) == 2:
+            name = str(err.loc_tuple()[0])
+        elif len(err.loc_tuple()) == 1:
+            name = str(err.loc_tuple()[0])
+        else:
+            name = "__all__"
+
+    if name in field_names:
+        return None
+
+    if message and not any(
+        [message.endswith("."), message.endswith("?"), message.endswith("!")]
+    ):
+        message = message + "."
+    message = message.capitalize()
+
+    return {"name": name, "message": message, "error_code": error_code}
+
+def raw_errors_to_fields(raw_errors: List) -> List[dict]:
+
+    fields = []
+    for top_err in raw_errors:
+        if hasattr(top_err.exc, "raw_errors"):
+            for err in top_err.exc.raw_errors:
+                # This is a special case when errors happen both in request
+                # handling & internal validation
+                if isinstance(err, list):
+                    err = err[0]
+                field_err = parse_error(
+                    err,
+                    field_names=list(map(lambda x: x["name"], fields)),
+                    raw=True,
+                )
+                if field_err is not None:
+                    fields.append(field_err)
+        else:
+            field_err = parse_error(
+                top_err,
+                field_names=list(map(lambda x: x["name"], fields)),
+                raw=False,
+            )
+            if field_err is not None:
+                fields.append(field_err)
+    return fields
+
+
+
+
+async def http_exception_handler(
+    request: Request, exc: StarletteHTTPException
+) -> UJSONResponse:
+    fields = getattr(exc, "fields", [])
+    message = getattr(exc, "detail", "Validation error.")
+    headers = getattr(exc, "headers", None)
+    if message and not any(
+        [message.endswith("."), message.endswith("?"), message.endswith("!")]
+    ):
+        message = message + "."
+    data = {
+        "error_codes": [getattr(exc, "error_code", exc.status_code)],
+        "message": message,
+        "fields": fields,
+    }
+    return UJSONResponse(data, status_code=exc.status_code, headers=headers)
+async def validation_exception_handler(
+    request: Request, exc: RequestValidationError
+) -> UJSONResponse:
+
+    status_code = getattr(exc, "status_code", 400)
+    headers = getattr(exc, "headers", None)
+    fields = raw_errors_to_fields(exc.raw_errors)
+
+    if fields:
+        error_codes = set(list(map(lambda x: x["error_code"], fields)))
+    else:
+        error_codes = [getattr(exc, "error_code", status_code)]
+
+    message = getattr(exc, "message", "Validation error.")
+    if message and not any(
+        [message.endswith("."), message.endswith("?"), message.endswith("!")]
+    ):
+        message = message + "."  # pragma: no cover
+
+    data = {"error_codes": error_codes, "message": message, "fields": fields}
+    return UJSONResponse(data, status_code=status_code, headers=headers)
+
+async def not_found_error_handler(
+    request: Request, exc: RequestValidationError
+) -> UJSONResponse:
+    code = getattr(exc, "error_code", 404)
+    detail = getattr(exc, "detail", "Not found.")
+    fields = getattr(exc, "fields", [])
+    headers = getattr(exc, "headers", None)
+    status_code = getattr(exc, "status_code", 404)
+    data = {"error_codes": [code], "message": detail, "fields": fields}
+    return UJSONResponse(data, status_code=status_code, headers=headers)
+
+async def internal_server_error_handler(
+    request: Request, exc: RequestValidationError
+) -> UJSONResponse:
+    code = getattr(exc, "error_code", 500)
+    detail = getattr(exc, "detail", "Internal Server Error.")
+    fields = getattr(exc, "fields", [])
+    headers = getattr(exc, "headers", None)
+    status_code = getattr(exc, "status_code", 500)
+    data = {"error_codes": [code], "message": detail, "fields": fields}
+    return UJSONResponse(data, status_code=status_code, headers=headers)
+
+
+
+[docs]
+def setup_exception_handlers(app: FastAPI) -> None:
+
+    app.add_exception_handler(StarletteHTTPException, http_exception_handler)
+    app.add_exception_handler(
+        RequestValidationError, validation_exception_handler
+    )
+    app.add_exception_handler(404, not_found_error_handler)
+    app.add_exception_handler(500, internal_server_error_handler)