Feat: Customise status code & Enhance exceptions (#100)

Feat: Customise status code & Enhance exceptions

.pre-commit-config.yaml

@@ -17,7 +17,7 @@ repos:
         - --py3-plus
         - --keep-runtime-typing
 -   repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: v0.0.255
+    rev: v0.0.262
     hooks:
     -   id: ruff
         args:
@@ -28,6 +28,6 @@ repos:
     -   id: isort
         name: isort (python)
 -   repo: https://github.com/psf/black
-    rev: 23.1.0
+    rev: 23.3.0
     hooks:
     -   id: black

fastapi_class/__init__.py

@@ -39,11 +39,7 @@ def post(self, user: ItemModel):
 
 __version__ = "3.1.0"
 
-from fastapi_class.exceptions import (
-    UNKNOWN_SERVER_ERROR_DETAIL,
-    ExceptionAbstract,
-    FormattedMessageException,
-)
+from fastapi_class.exception import FormattedMessageException
 from fastapi_class.openapi import ExceptionModel, _exceptions_to_responses
 from fastapi_class.routers import Metadata, Method, endpoint
 from fastapi_class.views import View
@@ -54,8 +50,6 @@ def post(self, user: ItemModel):
     "Method",
     "Metadata",
     "FormattedMessageException",
-    "UNKNOWN_SERVER_ERROR_DETAIL",
-    "ExceptionAbstract",
     "ExceptionModel",
     "_exceptions_to_responses",
 ]

fastapi_class/exception/__init__.py

@@ -0,0 +1,10 @@
+from fastapi_class.exception.handler import (
+    UNKOWN_SERVER_ERROR_DETAIL,
+    ExceptionAbstract,
+    FormattedMessageException,
+)
+
+__all__ = [
+    "FormattedMessageException" "ExceptionAbstract",
+    "UNKOWN_SERVER_ERROR_DETAIL",
+]

fastapi_class/exception/handler.py

@@ -0,0 +1,32 @@
+from abc import ABC, abstractmethod
+from collections.abc import Iterable
+from typing import Any
+
+from fastapi import HTTPException, status
+
+UNKOWN_SERVER_ERROR_DETAIL = "Unknown server error"
+
+
+class ExceptionAbstract(ABC):
+    _DEFAULT_DETAIL_SPECIAL_NAME = "__detail__"
+
+    def __init__(self, *, exceptions: Iterable[tuple[int, str]] | None = None) -> None:
+        self._exceptions = exceptions or [
+            (status.HTTP_500_INTERNAL_SERVER_ERROR, UNKOWN_SERVER_ERROR_DETAIL)
+        ]
+
+    @classmethod
+    @abstractmethod
+    def __call__(cls, *args: Any, **kwds: Any) -> Any:
+        raise NotImplementedError
+
+
+class FormattedMessageException(ExceptionAbstract):
+    def __call__(self, *_, **kwargs):
+        _exception = self._exceptions[0]
+
+        try:
+            detail = _exception[1].format(**kwargs)
+        except (IndexError, KeyError):
+            detail = _exception[1]
+        return HTTPException(status_code=_exception[0], detail=detail)

fastapi_class/logger.py

@@ -1,3 +1,3 @@
 import logging
 
-logging.getLogger("fastapi_class").addHandler(logging.NullHandler())
+logger = logging.getLogger("fastapi_class").addHandler(logging.NullHandler())

fastapi_class/routers.py

@@ -2,6 +2,7 @@
 from dataclasses import dataclass
 from enum import Enum
 from functools import wraps
+from typing import Any, ClassVar
 
 from fastapi.responses import Response
 from pydantic import BaseModel
@@ -20,15 +21,26 @@ class Metadata:
     methods: Iterable[Method]
     name: str | None = None
     path: str | None = None
+    status_code: int | None = None
     response_model: type[BaseModel] | None = None
     response_class: type[Response] | None = None
+    __default_method_suffix: ClassVar[str] = "_or_default"
+
+    def __getattr__(self, __name: str) -> Any | Callable[[Any], Any]:
+        if __name.endswith(Metadata.__default_method_suffix):
+            prefix = __name.removesuffix(Metadata.__default_method_suffix)
+            if hasattr(self, prefix):
+                return lambda _default: getattr(self, prefix, None) or _default
+            return getattr(self, prefix)
+        raise AttributeError(f"{self.__class__.__name__} has no attribute {__name}")
 
 
 def endpoint(
     methods: Iterable[str | Method] | None = None,
     *,
     name: str | None = None,
     path: str | None = None,
+    status_code: int | None = None,
     response_model: type[BaseModel] | None = None,
     response_class: type[Response] | None = None,
 ):
@@ -38,6 +50,7 @@ def endpoint(
     :param methods: methods
     :param name: name
     :param path: path
+    :param status_code: status code
     :param response_model: response model
     :param response_class: response class
 
@@ -90,6 +103,7 @@ async def _wrapper(*args, **kwargs):
             methods=parsed_method,
             name=name,
             path=path,
+            status_code=status_code,
             response_class=response_class,
             response_model=response_model,
         )

fastapi_class/views.py

@@ -1,7 +1,7 @@
 import re
 from collections.abc import Callable, Iterable
 
-from fastapi import APIRouter, FastAPI, HTTPException
+from fastapi import APIRouter, FastAPI, HTTPException, status
 from fastapi.responses import JSONResponse
 
 from fastapi_class.openapi import _exceptions_to_responses
@@ -20,17 +20,24 @@ def _view_class_name_default_parser(cls: object, method: str):
 EXCEPTIONS_ATTRIBUTE_NAME = "EXCEPTIONS"
 
 
+def _view_class_name_default_parser(cls: object, method: str):
+    class_name = " ".join(re.findall(r"[A-Z][^A-Z]*", cls.__name__.replace("View", "")))
+    return f"{method.capitalize()} {class_name}"
+
+
 def View(
     router: FastAPI | APIRouter,
     *,
     path: str = "/",
+    default_status_code: int = status.HTTP_200_OK,
     name_parser: Callable[[object, str], str] = _view_class_name_default_parser,
 ):
     """
     Class-based view decorator.
 
     :param router: router
     :param path: path
+    :param default_status_code: default status code
     :param name_parser: name parser
 
     :raise AssertionError: if router is not an instance of FastAPI or APIRouter
@@ -62,40 +69,31 @@ def _decorator(cls) -> None:
             if _callable_name in set(Method) or hasattr(
                 _callable, ENDPOINT_METADATA_ATTRIBUTE_NAME
             ):
-                metadata: Metadata | None = getattr(
-                    _callable, ENDPOINT_METADATA_ATTRIBUTE_NAME, None
-                )
-                response_model = (
-                    metadata.response_model
-                    if metadata and metadata.response_model
-                    else cls_based_response_model.get(_callable_name)
-                )
-                response_class = (
-                    metadata.response_class
-                    if metadata and metadata.response_class
-                    else cls_based_response_class.get(_callable_name, JSONResponse)
+                metadata: Metadata = getattr(
+                    _callable,
+                    ENDPOINT_METADATA_ATTRIBUTE_NAME,
+                    Metadata([_callable_name]),
                 )
                 exceptions: Iterable[HTTPException] = getattr(
                     obj, ENDPOINT_METADATA_ATTRIBUTE_NAME, {}
                 ).get(_callable_name, [])
                 exceptions += common_exceptions
-                method = list(metadata.methods) if metadata else [_callable_name]
-                name = (
-                    metadata.name
-                    if metadata and metadata.name
-                    else name_parser(cls, _callable_name)
-                )
                 _path = path
                 if metadata and metadata.path:
                     _path = path + metadata.path
                 router.add_api_route(
                     _path,
                     _callable,
-                    methods=method,
-                    response_class=response_class,
-                    response_model=response_model,
+                    methods=list(metadata.methods),
+                    response_class=metadata.response_class_or_default(
+                        cls_based_response_class.get(_callable_name, JSONResponse)
+                    ),
+                    response_model=metadata.response_model_or_default(
+                        cls_based_response_model.get(_callable_name)
+                    ),
                     responses=_exceptions_to_responses(exceptions),
-                    name=name,
+                    name=metadata.name_or_default(name_parser(cls, _callable_name)),
+                    status_code=metadata.status_code_or_default(default_status_code),
                 )
 
     return _decorator

pyproject.toml

@@ -71,6 +71,7 @@ ignore = [
     "E501",  # line too long, handled by black
     "B008",  # do not perform function calls in argument defaults
     "C901",  # too complex
+    "B018",  # Found useless expression
 ]
 
 [tool.ruff.per-file-ignores]

tests/factory.py

@@ -0,0 +1,8 @@
+from typing import Any, Generic, TypeVar
+
+T = TypeVar("T")
+
+
+class Factory(Generic[T]):
+    def __call__(self, *args: Any, **kwds: Any) -> T:
+        pass  # pragma: no cover

tests/test_exceptions.py

@@ -2,18 +2,18 @@
 
 import pytest
 
-from fastapi_class import (
-    UNKNOWN_SERVER_ERROR_DETAIL,
+from fastapi_class.exception import (
+    UNKOWN_SERVER_ERROR_DETAIL,
     ExceptionAbstract,
     FormattedMessageException,
 )
 
 
-@patch("fastapi_class.ExceptionAbstract.__abstractmethods__", set())
+@patch("fastapi_class.exception.ExceptionAbstract.__abstractmethods__", set())
 def test_abstract_factory_creation__defaults():
     _instance = ExceptionAbstract()
-    assert _instance.exceptions[0][0] == 500
-    assert _instance.exceptions[0][1] == UNKNOWN_SERVER_ERROR_DETAIL
+    assert _instance._exceptions[0][0] == 500
+    assert _instance._exceptions[0][1] == UNKOWN_SERVER_ERROR_DETAIL
 
 
 @pytest.mark.parametrize("keyword_args", ({}, {"some_var": 0}))
@@ -26,3 +26,9 @@ def test_formatted_message_factory__non_formattable_string(keyword_args: dict):
 def test_formatted_message_factory__format_string(arg: str | int | float | bool):
     _instance = FormattedMessageException(exceptions=((500, "Test {test}"),))
     assert _instance(**{"test": arg}).detail == f"Test {arg}"
+
+
+@patch("fastapi_class.exception.ExceptionAbstract.__abstractmethods__", set())
+def test_formatted_message_factory__format_string__missing_key():
+    _instance = FormattedMessageException(exceptions=((500, "Test {test}"),))
+    assert _instance().detail == "Test {test}"

tests/test_logger.py

@@ -0,0 +1,13 @@
+import pytest
+
+from fastapi_class import logger
+
+
+@pytest.fixture
+def setup_logger():
+    return logger
+
+
+def test_logger_disable(setup_logger):
+    logger = setup_logger
+    logger.disabled = True

tests/test_metadata.py

@@ -0,0 +1,63 @@
+from typing import Any
+
+import pytest
+from fastapi.responses import Response
+from pydantic import BaseModel
+
+from fastapi_class import Metadata
+from tests.factory import Factory
+
+
+@pytest.fixture(name="metadata_factory")
+def fixture_metadata_factory():
+    def _factory(data: dict[str, Any] | None = None):
+        data = data or {}
+        return Metadata(
+            data.get("methods", ["GET"]),
+            name=data.get("name", "test"),
+            path=data.get("path", "test"),
+            status_code=data.get("status_code", 200),
+            response_model=data.get("response_model", BaseModel),
+            response_class=data.get("response_class", Response),
+        )
+
+    return _factory
+
+
+@pytest.fixture(name="metadata")
+def fixture_metadata(metadata_factory: Factory[Metadata]):
+    return metadata_factory()
+
+
+def test_metadata_dynamic_optional_fields__get_when_field_exist(metadata: Metadata):
+    assert metadata.methods == ["GET"]
+    assert metadata.name == "test"
+    assert metadata.status_code == 200
+    assert metadata.response_model == BaseModel
+    assert metadata.response_class == Response
+
+
+def test_metadata_dynamic_optional_fields__raise_when_field_doesnt_exist(
+    metadata: Metadata,
+):
+    with pytest.raises(AttributeError):
+        metadata.non_existing
+    with pytest.raises(AttributeError):
+        metadata.or_default
+
+
+def test_metadata_dynamic_optional_fields_default__when_field_exists(
+    metadata_factory: Factory[Metadata],
+):
+    metadata = metadata_factory({"name": "", "methods": [], "status_code": None})
+
+    assert metadata.name_or_default("test") == "test"
+    assert metadata.methods_or_default(["test"]) == ["test"]
+    assert metadata.status_code_or_default(123) == 123
+
+
+def test_metadata_dynamic_optional_fields_default__when_field_doesnt_exist(
+    metadata: Metadata,
+):
+    with pytest.raises(AttributeError):
+        metadata.non_existing_or_default("test")