Error Handling

How to create your own error types.

Dispatch automatically handles execution failures and retries depending on how your code will fail. Under the hood, the Dispatch scheduler is leveraging an adaptive concurrency algorithm so you don't have to think about retries ever again.

The scheduler will try to do the right thing, but in some cases, you might need to teach it what the right thing is. To that end, you can create custom error types or wrap well-known errors.

Handle exceptions

You can first define how the system should handle exceptions. Say you want EOFError exceptions to fail permanently (by default, we treat EOF errors as retriable):

from dispatch.status import Status, register_error_type, status_for_error

def custom_exception_handler(error: Exception) -> Status:
    if error is EOFError:
        return Status.PERMANENT_ERROR

register_error_type(EOFError, custom_exception_handler)

Wrap and customize well-known errors

As an example, let's take a look at how the Github API manages rate limits:

If you exceed your primary rate limit, you will receive a 403 or 429 response, and the x-ratelimit-remaining header will be 0.

Since the Github API might return a 403 during rate limit events (usually used to refuse to authorize requests), we need to wrap those errors to retry the request:

from dispatch.status import Status, register_error_type
from dispatch.integrations.httpx import httpx_error_status

# Handle the Github-specific cases.
def github_error_status(error: Exception) -> Status:
    match error:
        case httpx.HTTPStatusError():
            resp = error.response
            X_RateLimit_Remaining = resp.headers.get("X-RateLimit-Remaining")
            if resp.status_code == 403 and X_RateLimit_Remaining == "0":
                return Status.TEMPORARY_ERROR
    # Use the well-known errors for all the other cases.
    return httpx_error_status(error)

# Register your customer error handler.
register_error_type(httpx.HTTPError, github_error_status)

You can find integrations with well-known errors in the integrations module. See:

Create custom error types

What if you want to instruct Dispatch to retry a function based on its result? Let's look at a simple example.

class Result:
    def __init__(self, success: bool):
        self.success = success

def processing(args) -> Result
    // ...
    return Result(success=false)

To create a custom error type, you need to:

  • Create a handler for your custom Response type.

  • Register the handler to let Dispatch make the right scheduling decision.

from dispatch.status import Status, register_error_type, register_output_type

# Create a handler for your Response type
def handler_response(response: Response) -> Status:
    if not response.success:
        return Status.TEMPORARY_ERROR
    return Status.PERMANENT_ERROR

# Register the response handler
register_output_type(Response, handler_response)

Last updated

©️ Stealth Rocket, Inc. All rights reserved.