Merge pull request #168 from canvas-medical/csande/exceptions

Added exception classes for most status codes mentioned in the FHIR spec

fhirstarter/exceptions.py

@@ -1,8 +1,7 @@
 """
 Standard exception types for reporting errors.
 
-The exception classes defined here provide a response method which will return a Response
-containing an OperationOutcome and an HTTP status code.
+The exception classes defined here provide a method which will return an OperationOutcome.
 """
 
 from abc import ABC, abstractmethod
@@ -34,6 +33,42 @@ def operation_outcome(self) -> OperationOutcome:
         raise NotImplementedError
 
 
+class FHIRHTTPException(FHIRException):
+    """
+    Abstract base class for FHIR errors that map neatly to OperationOutcome issue type codes.
+    Example: HTTP status code 409 <-> OperationOutcome issue type code "conflict"
+
+    If these mappings are not desired, FHIRGeneralError can be used to create the exact desired
+    response.
+    """
+
+    _STATUS_CODE_MAPPINGS = {
+        401: "unknown",
+        403: "forbidden",
+        405: "not-supported",
+        406: "not-supported",
+        409: "conflict",
+        410: "deleted",
+        412: "conflict",
+        415: "not-supported",
+    }
+
+    def __init__(self, details_text: str | None = None):
+        super().__init__(self._status_code(), details_text)
+
+    def operation_outcome(self) -> OperationOutcome:
+        return make_operation_outcome(
+            severity="error",
+            code=self._STATUS_CODE_MAPPINGS[self._status_code()],
+            details_text=self.detail,
+        )
+
+    @classmethod
+    @abstractmethod
+    def _status_code(cls) -> int:
+        raise NotImplementedError
+
+
 class FHIRGeneralError(FHIRException):
     """
     General FHIR exception class.
@@ -72,28 +107,20 @@ def operation_outcome(self) -> OperationOutcome:
         )
 
 
-class FHIRUnauthorizedError(FHIRException):
-    """FHIR exception class for 401 authentication errors."""
+class FHIRUnauthorizedError(FHIRHTTPException):
+    """FHIR exception class for 401 unauthorized errors."""
 
-    def __init__(self, details_text: str) -> None:
-        super().__init__(status.HTTP_401_UNAUTHORIZED, details_text)
-
-    def operation_outcome(self) -> OperationOutcome:
-        return make_operation_outcome(
-            severity="error", code="unknown", details_text=self.detail
-        )
+    @classmethod
+    def _status_code(cls) -> int:
+        return status.HTTP_401_UNAUTHORIZED
 
 
-class FHIRForbiddenError(FHIRException):
+class FHIRForbiddenError(FHIRHTTPException):
     """FHIR exception class for 403 forbidden errors."""
 
-    def __init__(self, details_text: str) -> None:
-        super().__init__(status.HTTP_403_FORBIDDEN, details_text)
-
-    def operation_outcome(self) -> OperationOutcome:
-        return make_operation_outcome(
-            severity="error", code="forbidden", details_text=self.detail
-        )
+    @classmethod
+    def _status_code(cls) -> int:
+        return status.HTTP_403_FORBIDDEN
 
 
 class FHIRResourceNotFoundError(FHIRException):
@@ -117,3 +144,64 @@ def operation_outcome(self) -> OperationOutcome:
                 f"Unknown {interaction_info.resource_type} resource "
                 f"'{interaction_info.resource_id}'",
             )
+
+
+class FHIRMethodNotAllowedError(FHIRHTTPException):
+    """FHIR exception class for 405 method not allowed errors."""
+
+    @classmethod
+    def _status_code(cls) -> int:
+        return status.HTTP_405_METHOD_NOT_ALLOWED
+
+
+class FHIRNotAcceptableError(FHIRHTTPException):
+    """FHIR exception class for 406 not acceptable errors."""
+
+    @classmethod
+    def _status_code(cls) -> int:
+        return status.HTTP_406_NOT_ACCEPTABLE
+
+
+class FHIRConflictError(FHIRHTTPException):
+    """FHIR exception class for 409 conflict errors."""
+
+    @classmethod
+    def _status_code(cls) -> int:
+        return status.HTTP_409_CONFLICT
+
+
+class FHIRGoneError(FHIRHTTPException):
+    """FHIR exception class for 410 gone errors."""
+
+    @classmethod
+    def _status_code(cls) -> int:
+        return status.HTTP_410_GONE
+
+
+class FHIRPreconditionFailedError(FHIRHTTPException):
+    """FHIR exception class for 412 precondition failed errors."""
+
+    @classmethod
+    def _status_code(cls) -> int:
+        return status.HTTP_412_PRECONDITION_FAILED
+
+
+class FHIRUnsupportedMediaTypeError(FHIRHTTPException):
+    """FHIR exception class for 415 unsupported media type errors."""
+
+    @classmethod
+    def _status_code(cls) -> int:
+        return status.HTTP_415_UNSUPPORTED_MEDIA_TYPE
+
+
+class FHIRUnprocessableEntityError(FHIRException):
+    """FHIR exception class for 422 unprocessable entity errors."""
+
+    def __init__(self, code: str, details_text: str) -> None:
+        super().__init__(status.HTTP_422_UNPROCESSABLE_ENTITY, details_text)
+        self._code = code
+
+    def operation_outcome(self) -> OperationOutcome:
+        return make_operation_outcome(
+            severity="error", code=self._code, details_text=self.detail
+        )

fhirstarter/tests/test_errors.py

@@ -7,6 +7,18 @@
 import pytest
 from fastapi import HTTPException
 
+from ..exceptions import (
+    FHIRBadRequestError,
+    FHIRConflictError,
+    FHIRForbiddenError,
+    FHIRGoneError,
+    FHIRMethodNotAllowedError,
+    FHIRNotAcceptableError,
+    FHIRPreconditionFailedError,
+    FHIRUnauthorizedError,
+    FHIRUnprocessableEntityError,
+    FHIRUnsupportedMediaTypeError,
+)
 from ..fhirstarter import FHIRProvider, FHIRStarter, Request, Response, status
 from ..testclient import TestClient
 from ..utils import make_operation_outcome
@@ -122,33 +134,160 @@ def test_validation_error(
     )
 
 
-def _handler_exception_async() -> Callable[..., Coroutine[None, None, Patient]]:
+def _handler_exception_async(
+    exception: HTTPException,
+) -> Callable[..., Coroutine[None, None, Patient]]:
     """Return an async Patient read handler."""
 
     async def patient_read(*_: Any) -> Patient:
-        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)
+        raise exception
 
     return patient_read
 
 
-def _handler_exception() -> Callable[..., Patient]:
+def _handler_exception(exception: HTTPException) -> Callable[..., Patient]:
     """Return a Patient read handler."""
 
     def patient_read(*_: Any) -> Patient:
-        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)
+        raise exception
 
     return patient_read
 
 
 @pytest.mark.parametrize(
-    argnames="handler",
-    argvalues=[_handler_exception_async(), _handler_exception()],
+    argnames="exception,status_code,issue",
+    argvalues=[
+        (
+            HTTPException(status_code=status.HTTP_400_BAD_REQUEST),
+            status.HTTP_400_BAD_REQUEST,
+            {
+                "severity": "error",
+                "code": "processing",
+                "details": {"text": "Bad Request"},
+            },
+        ),
+        (
+            FHIRBadRequestError(code="processing", details_text="Error"),
+            status.HTTP_400_BAD_REQUEST,
+            {
+                "severity": "error",
+                "code": "processing",
+                "details": {"text": "Error"},
+            },
+        ),
+        (
+            FHIRUnauthorizedError(details_text="Error"),
+            status.HTTP_401_UNAUTHORIZED,
+            {
+                "severity": "error",
+                "code": "unknown",
+                "details": {"text": "Error"},
+            },
+        ),
+        (
+            FHIRForbiddenError(details_text="Error"),
+            status.HTTP_403_FORBIDDEN,
+            {
+                "severity": "error",
+                "code": "forbidden",
+                "details": {"text": "Error"},
+            },
+        ),
+        (
+            FHIRMethodNotAllowedError(details_text="Error"),
+            status.HTTP_405_METHOD_NOT_ALLOWED,
+            {
+                "severity": "error",
+                "code": "not-supported",
+                "details": {"text": "Error"},
+            },
+        ),
+        (
+            FHIRNotAcceptableError(details_text="Error"),
+            status.HTTP_406_NOT_ACCEPTABLE,
+            {
+                "severity": "error",
+                "code": "not-supported",
+                "details": {"text": "Error"},
+            },
+        ),
+        (
+            FHIRConflictError(details_text="Error"),
+            status.HTTP_409_CONFLICT,
+            {
+                "severity": "error",
+                "code": "conflict",
+                "details": {"text": "Error"},
+            },
+        ),
+        (
+            FHIRGoneError(details_text="Error"),
+            status.HTTP_410_GONE,
+            {
+                "severity": "error",
+                "code": "deleted",
+                "details": {"text": "Error"},
+            },
+        ),
+        (
+            FHIRPreconditionFailedError(details_text="Error"),
+            status.HTTP_412_PRECONDITION_FAILED,
+            {
+                "severity": "error",
+                "code": "conflict",
+                "details": {"text": "Error"},
+            },
+        ),
+        (
+            FHIRUnsupportedMediaTypeError(details_text="Error"),
+            status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
+            {
+                "severity": "error",
+                "code": "not-supported",
+                "details": {"text": "Error"},
+            },
+        ),
+        (
+            FHIRUnprocessableEntityError(code="invalid", details_text="Error"),
+            status.HTTP_422_UNPROCESSABLE_ENTITY,
+            {
+                "severity": "error",
+                "code": "invalid",
+                "details": {"text": "Error"},
+            },
+        ),
+    ],
+    ids=[
+        "http",
+        "bad request",
+        "unauthorized",
+        "forbidden",
+        "method not allowed",
+        "not acceptable",
+        "conflict",
+        "gone",
+        "precondition failed",
+        "unsupported media type",
+        "unprocessable entity",
+    ],
+)
+@pytest.mark.parametrize(
+    argnames="handler_func",
+    argvalues=[_handler_exception_async, _handler_exception],
     ids=["async", "nonasync"],
 )
-def test_http_exception(
-    handler: Callable[..., Coroutine[None, None, Patient]] | Callable[..., Patient]
+def test_exception(
+    exception: HTTPException,
+    status_code: int,
+    issue: Mapping[str, Any],
+    handler_func: Callable[
+        [HTTPException],
+        Callable[..., Coroutine[None, None, Patient]] | Callable[..., Patient],
+    ],
 ) -> None:
-    """Test exception handling for HTTP Exceptions."""
+    """Test exception handling for HTTP and FHIR exceptions."""
+    handler = handler_func(exception)
+
     provider = FHIRProvider()
     provider.read(Patient)(handler)
 
@@ -158,16 +297,10 @@ def test_http_exception(
 
     assert_expected_response(
         response,
-        status.HTTP_400_BAD_REQUEST,
+        status_code,
         content={
             "resourceType": "OperationOutcome",
-            "issue": [
-                {
-                    "severity": "error",
-                    "code": "processing",
-                    "details": {"text": "Bad Request"},
-                }
-            ],
+            "issue": [issue],
         },
     )
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "fhirstarter"
-version = "1.4.2"
+version = "1.5.0"
 description = "An ASGI FHIR API framework built on top of FastAPI and FHIR Resources"
 authors = ["Christopher Sande <christopher.sande@canvasmedical.com>"]
 maintainers = ["Canvas Medical Engineering <engineering@canvasmedical.com>"]