Migrate to fastapi-error-map 1.0.0

Author: ivan-borovets

pyproject.toml

@@ -13,7 +13,7 @@ dependencies = [
   "bcrypt==5.0.0",
   "dishka==1.10.1",
   "fastapi==0.136.1",
-  "fastapi-error-map==0.9.10",
+  "fastapi-error-map==1.0.0",
   "psycopg[binary]==3.3.4",
   "pydantic-settings==2.14.1",
   "pyjwt[crypto]==2.12.1",
@@ -256,8 +256,10 @@ force-wrap-aliases = true
 split-on-trailing-comma = true
 
 [tool.ruff.lint.per-file-ignores]
-"scripts/**" = ["INP001"]  # runnable scripts, not an importable package
-"scripts/dishka/plot_dependencies_data.py" = ["T201"]  # print
+"scripts/**" = [
+  "INP001",  # runnable scripts, not an importable package
+  "T20",  # print is legitimate output here
+]
 "src/app/outbound/persistence_sqla/alembic/**" = ["ALL"]
 "tests/**" = [
   "ARG002",  # unused-method-argument

src/app/inbound/http/account/change_password.py

@@ -4,12 +4,12 @@
 from dishka.integrations.fastapi import inject
 from fastapi import APIRouter, Depends, status
 from fastapi.security import APIKeyCookie
-from fastapi_error_map import ErrorAwareRouter
 from pydantic import BaseModel, ConfigDict
 
 from app.core.common.authorization.exceptions import AuthorizationError
 from app.core.common.exceptions import BusinessTypeError
 from app.inbound.http.errors.callbacks import log_info
+from app.inbound.http.errors.router import make_error_aware_router
 from app.inbound.http.errors.rules import HTTP_503_SERVICE_UNAVAILABLE_RULE
 from app.outbound.adapters.exceptions import PasswordHasherBusyError
 from app.outbound.auth_ctx.exceptions import AuthenticationChangeError, AuthenticationError, ReAuthenticationError
@@ -30,7 +30,7 @@ class ChangePasswordRequestSchema(BaseModel):
 
 
 def make_change_password_router(*, cookie_name: str) -> APIRouter:
-    router = ErrorAwareRouter()
+    router = make_error_aware_router(on_error=log_info)
 
     @router.put(
         "/password/",
@@ -43,7 +43,6 @@ def make_change_password_router(*, cookie_name: str) -> APIRouter:
             ReAuthenticationError: status.HTTP_403_FORBIDDEN,
             PasswordHasherBusyError: HTTP_503_SERVICE_UNAVAILABLE_RULE,
         },
-        default_on_error=log_info,
         status_code=status.HTTP_204_NO_CONTENT,
         dependencies=[Depends(APIKeyCookie(name=cookie_name))],
         description=getdoc(ChangePassword),

src/app/inbound/http/account/log_in.py

@@ -3,11 +3,11 @@
 from dishka import FromDishka
 from dishka.integrations.fastapi import inject
 from fastapi import APIRouter, status
-from fastapi_error_map import ErrorAwareRouter
 
 from app.core.common.authorization.exceptions import AuthorizationError
 from app.core.common.exceptions import BusinessTypeError
 from app.inbound.http.errors.callbacks import log_info
+from app.inbound.http.errors.router import make_error_aware_router
 from app.inbound.http.errors.rules import HTTP_503_SERVICE_UNAVAILABLE_RULE
 from app.outbound.adapters.exceptions import PasswordHasherBusyError
 from app.outbound.auth_ctx.exceptions import AlreadyAuthenticatedError, AuthenticationError
@@ -16,7 +16,7 @@
 
 
 def make_log_in_router() -> APIRouter:
-    router = ErrorAwareRouter()
+    router = make_error_aware_router(on_error=log_info)
 
     @router.post(
         "/login/",
@@ -28,7 +28,6 @@ def make_log_in_router() -> APIRouter:
             AuthenticationError: status.HTTP_401_UNAUTHORIZED,
             PasswordHasherBusyError: HTTP_503_SERVICE_UNAVAILABLE_RULE,
         },
-        default_on_error=log_info,
         status_code=status.HTTP_204_NO_CONTENT,
         description=getdoc(LogIn),
     )

src/app/inbound/http/account/log_out.py

@@ -4,18 +4,18 @@
 from dishka.integrations.fastapi import inject
 from fastapi import APIRouter, Depends, status
 from fastapi.security import APIKeyCookie
-from fastapi_error_map import ErrorAwareRouter
 
 from app.core.common.authorization.exceptions import AuthorizationError
 from app.inbound.http.errors.callbacks import log_info
+from app.inbound.http.errors.router import make_error_aware_router
 from app.inbound.http.errors.rules import HTTP_503_SERVICE_UNAVAILABLE_RULE
 from app.outbound.auth_ctx.exceptions import AuthenticationError
 from app.outbound.auth_ctx.handlers.log_out import LogOut
 from app.outbound.exceptions import StorageError
 
 
 def make_log_out_router(*, cookie_name: str) -> APIRouter:
-    router = ErrorAwareRouter()
+    router = make_error_aware_router(on_error=log_info)
 
     @router.delete(
         "/logout/",
@@ -24,7 +24,6 @@ def make_log_out_router(*, cookie_name: str) -> APIRouter:
             StorageError: HTTP_503_SERVICE_UNAVAILABLE_RULE,
             AuthorizationError: status.HTTP_403_FORBIDDEN,
         },
-        default_on_error=log_info,
         status_code=status.HTTP_204_NO_CONTENT,
         dependencies=[Depends(APIKeyCookie(name=cookie_name))],
         description=getdoc(LogOut),

src/app/inbound/http/account/sign_up.py

@@ -3,12 +3,12 @@
 from dishka import FromDishka
 from dishka.integrations.fastapi import inject
 from fastapi import APIRouter, status
-from fastapi_error_map import ErrorAwareRouter
 
 from app.core.commands.exceptions import UsernameAlreadyExistsError
 from app.core.common.authorization.exceptions import AuthorizationError
 from app.core.common.exceptions import BusinessTypeError
 from app.inbound.http.errors.callbacks import log_info
+from app.inbound.http.errors.router import make_error_aware_router
 from app.inbound.http.errors.rules import HTTP_503_SERVICE_UNAVAILABLE_RULE
 from app.outbound.adapters.exceptions import PasswordHasherBusyError
 from app.outbound.auth_ctx.exceptions import AlreadyAuthenticatedError
@@ -17,7 +17,7 @@
 
 
 def make_sign_up_router() -> APIRouter:
-    router = ErrorAwareRouter()
+    router = make_error_aware_router(on_error=log_info)
 
     @router.post(
         "/signup/",
@@ -29,7 +29,6 @@ def make_sign_up_router() -> APIRouter:
             PasswordHasherBusyError: HTTP_503_SERVICE_UNAVAILABLE_RULE,
             UsernameAlreadyExistsError: status.HTTP_409_CONFLICT,
         },
-        default_on_error=log_info,
         status_code=status.HTTP_204_NO_CONTENT,
         description=getdoc(SignUp),
     )

src/app/inbound/http/api_v1_router.py

@@ -1,11 +1,12 @@
 from fastapi import APIRouter
 
 from app.inbound.http.account.router import make_account_router
+from app.inbound.http.errors.openapi_responses import SERVER_ERROR_RESPONSES
 from app.inbound.http.users.router import make_users_router
 
 
 def make_v1_router(*, cookie_name: str) -> APIRouter:
-    router = APIRouter(prefix="/api/v1")
+    router = APIRouter(prefix="/api/v1", responses=SERVER_ERROR_RESPONSES)
     router.include_router(make_account_router(cookie_name=cookie_name))
     router.include_router(make_users_router(cookie_name=cookie_name))
     return router

src/app/inbound/http/errors/internal_server_error.py

@@ -0,0 +1,7 @@
+from fastapi_error_map import StructuredErrorResponse, Translator, structured
+from starlette import status
+
+
+def internal_server_error(exc: Exception) -> StructuredErrorResponse:
+    translate: Translator[StructuredErrorResponse] = structured()(status.HTTP_500_INTERNAL_SERVER_ERROR)
+    return translate(exc)

src/app/inbound/http/errors/openapi_responses.py

@@ -0,0 +1,10 @@
+from typing import Any, Final
+
+from fastapi_error_map import StructuredErrorResponse
+from starlette import status
+
+type OpenApiResponses = dict[int | str, dict[str, Any]]
+
+SERVER_ERROR_RESPONSES: Final[OpenApiResponses] = {
+    status.HTTP_500_INTERNAL_SERVER_ERROR: {"model": StructuredErrorResponse},
+}

src/app/inbound/http/errors/router.py

@@ -0,0 +1,19 @@
+from typing import Any
+
+from fastapi_error_map import ErrorAwareRouter, ErrorMap, OnError, structured
+
+
+def make_error_aware_router(
+    *,
+    error_map: ErrorMap | None = None,
+    on_error: OnError | None = None,
+    warn_on_unmapped: bool = True,
+    **kwargs: Any,
+) -> ErrorAwareRouter:
+    return ErrorAwareRouter(
+        translator_factory=structured(),
+        error_map=error_map,
+        on_error=on_error,
+        warn_on_unmapped=warn_on_unmapped,
+        **kwargs,
+    )

src/app/inbound/http/errors/rules.py

@@ -1,11 +1,11 @@
 from typing import Final
 
-from fastapi_error_map.rules import Rule, rule
+from fastapi_error_map import Rule, rule, structured
 from starlette import status
 
-from app.inbound.http.errors.translators import ServiceUnavailableTranslator
+SERVICE_UNAVAILABLE_MESSAGE: Final[str] = "Service temporarily unavailable. Please try again later."
 
 HTTP_503_SERVICE_UNAVAILABLE_RULE: Final[Rule] = rule(
     status=status.HTTP_503_SERVICE_UNAVAILABLE,
-    translator=ServiceUnavailableTranslator(),
+    translator=structured(server_message=SERVICE_UNAVAILABLE_MESSAGE)(status.HTTP_503_SERVICE_UNAVAILABLE),
 )