Add dedicated error types for each of the error statuses

examples/github_stats/app.py

@@ -17,6 +17,7 @@
 import httpx
 from fastapi import FastAPI
 
+from dispatch.error import ThrottleError
 from dispatch.fastapi import Dispatch
 
 app = FastAPI()
@@ -29,7 +30,7 @@ def get_gh_api(url):
     response = httpx.get(url)
     X_RateLimit_Remaining = response.headers.get("X-RateLimit-Remaining")
     if response.status_code == 403 and X_RateLimit_Remaining == "0":
-        raise EOFError("Rate limit exceeded")
+        raise ThrottleError("Rate limit exceeded")
     response.raise_for_status()
     return response.json()
 

src/dispatch/error.py

@@ -1,3 +1,105 @@
-class IncompatibleStateError(Exception):
+from builtins import TimeoutError as _TimeoutError
+from typing import cast
+
+from dispatch.status import Status, register_error_type
+
+
+class DispatchError(Exception):
+    """Base class for Dispatch exceptions."""
+
+    _status = Status.UNSPECIFIED
+
+
+class TimeoutError(DispatchError, _TimeoutError):
+    """Operation timed out."""
+
+    _status = Status.TIMEOUT
+
+
+class ThrottleError(DispatchError):
+    """Operation was throttled."""
+
+    _status = Status.THROTTLED
+
+
+class InvalidArgumentError(DispatchError, ValueError):
+    """Invalid argument was received."""
+
+    _status = Status.INVALID_ARGUMENT
+
+
+class InvalidResponseError(DispatchError, ValueError):
+    """Invalid response was received."""
+
+    _status = Status.INVALID_RESPONSE
+
+
+class TemporaryError(DispatchError):
+    """Generic temporary error. Used in cases where a more specific
+    error class is not available, but the operation that failed should
+    be attempted again."""
+
+    _status = Status.TEMPORARY_ERROR
+
+
+class PermanentError(DispatchError):
+    """Generic permanent error. Used in cases where a more specific
+    error class is not available, but the operation that failed should
+    *not* be attempted again."""
+
+    _status = Status.PERMANENT_ERROR
+
+
+class IncompatibleStateError(DispatchError):
     """Coroutine state is incompatible with the current interpreter
     and application revision."""
+
+    _status = Status.INCOMPATIBLE_STATE
+
+
+class DNSError(DispatchError, ConnectionError):
+    """Generic DNS error. Used in cases where a more specific error class is
+    not available, but the operation that failed should be attempted again."""
+
+    _status = Status.DNS_ERROR
+
+
+class TCPError(DispatchError, ConnectionError):
+    """Generic TCP error. Used in cases where a more specific error class is
+    not available, but the operation that failed should be attempted again."""
+
+    _status = Status.TCP_ERROR
+
+
+class HTTPError(DispatchError, ConnectionError):
+    """Generic HTTP error. Used in cases where a more specific error class is
+    not available, but the operation that failed should be attempted again."""
+
+    _status = Status.HTTP_ERROR
+
+
+class UnauthenticatedError(DispatchError):
+    """The caller did not authenticate with the resource."""
+
+    _status = Status.UNAUTHENTICATED
+
+
+class PermissionDeniedError(DispatchError, PermissionError):
+    """The caller does not have access to the resource."""
+
+    _status = Status.PERMISSION_DENIED
+
+
+class NotFoundError(DispatchError):
+    """Generic not found error. Used in cases where a more specific error class
+    is not available, but the operation that failed should *not* be attempted
+    again."""
+
+    _status = Status.NOT_FOUND
+
+
+def dispatch_error_status(error: Exception) -> Status:
+    return cast(DispatchError, error)._status
+
+
+register_error_type(DispatchError, dispatch_error_status)

src/dispatch/status.py

@@ -1,7 +1,6 @@
 import enum
 from typing import Any, Callable, Dict, Type
 
-from dispatch.error import IncompatibleStateError
 from dispatch.sdk.v1 import status_pb2 as status_pb
 
 
@@ -92,9 +91,7 @@ def status_for_error(error: BaseException) -> Status:
     # If not, resort to standard error categorization.
     #
     # See https://docs.python.org/3/library/exceptions.html
-    if isinstance(error, IncompatibleStateError):
-        return Status.INCOMPATIBLE_STATE
-    elif isinstance(error, TimeoutError):
+    if isinstance(error, TimeoutError):
         return Status.TIMEOUT
     elif isinstance(error, TypeError) or isinstance(error, ValueError):
         return Status.INVALID_ARGUMENT

tests/dispatch/test_status.py

@@ -1,5 +1,6 @@
 import unittest
 
+from dispatch import error
 from dispatch.integrations.http import http_response_code_status
 from dispatch.status import Status, status_for_error
 
@@ -68,6 +69,27 @@ class CustomError(TimeoutError):
 
         assert status_for_error(CustomError()) is Status.TIMEOUT
 
+    def test_status_for_DispatchError(self):
+        assert status_for_error(error.TimeoutError()) is Status.TIMEOUT
+        assert status_for_error(error.ThrottleError()) is Status.THROTTLED
+        assert status_for_error(error.InvalidArgumentError()) is Status.INVALID_ARGUMENT
+        assert status_for_error(error.InvalidResponseError()) is Status.INVALID_RESPONSE
+        assert status_for_error(error.TemporaryError()) is Status.TEMPORARY_ERROR
+        assert status_for_error(error.PermanentError()) is Status.PERMANENT_ERROR
+        assert (
+            status_for_error(error.IncompatibleStateError())
+            is Status.INCOMPATIBLE_STATE
+        )
+        assert status_for_error(error.DNSError()) is Status.DNS_ERROR
+        assert status_for_error(error.TCPError()) is Status.TCP_ERROR
+        assert status_for_error(error.HTTPError()) is Status.HTTP_ERROR
+        assert status_for_error(error.UnauthenticatedError()) is Status.UNAUTHENTICATED
+        assert (
+            status_for_error(error.PermissionDeniedError()) is Status.PERMISSION_DENIED
+        )
+        assert status_for_error(error.NotFoundError()) is Status.NOT_FOUND
+        assert status_for_error(error.DispatchError()) is Status.UNSPECIFIED
+
 
 class TestHTTPStatusCodes(unittest.TestCase):
     def test_http_response_code_status_400(self):