Exception Handling

ZodiacCore provides a centralized exception handling system that automatically converts Python exceptions into standardized, production-ready JSON responses.

1. Core Concepts

The ZodiacException Base

Business logic errors should inherit from ZodiacException directly or from one of the built-in exception families. This base class allows you to define:

• http_code: The HTTP status code (e.g., 404, 400).
• code: A custom business error code.
• message: A human-readable error description.
• data: Optional payload for additional error details (e.g., validation errors).

http_code controls the HTTP response status. code is the business error code in the response body; it can differ from the HTTP status.

Automatic Transformation

When a ZodiacException is raised, the handler_zodiac_exception exception handler catches it and transforms it into a standard JSON response:

{
  "code": 404,
  "message": "Resource not found",
  "data": null
}

---

2. Validation Errors (HTTP 422)

One of the best features of ZodiacCore is that it also standardizes framework-level validation errors. When a user sends invalid JSON or missing parameters, FastAPI normally returns a custom structure. ZodiacCore catches these and wraps them in our standard format:

{
  "code": 422,
  "message": "Unprocessable Entity",
  "data": [
    {
      "type": "missing",
      "loc": ["body", "username"],
      "msg": "Field required",
      "input": null
    }
  ]
}

This ensures your API is 100% consistent, whether the error came from your business logic or from a schema mismatch.

---

3. Built-in Exceptions

ZodiacCore includes several common exceptions ready to use:

Exception	HTTP Status	Use Case
BadRequestException	400	Invalid input or parameters.
UpstreamServiceException	400	A third-party/upstream service failed, timed out, or returned an unexpected status.
UpstreamRequestException	400	The upstream service rejected this service's request, usually HTTP 400 or 422.
UnauthorizedException	401	Missing or invalid authentication.
ForbiddenException	403	Insufficient permissions.
NotFoundException	404	Resource does not exist.
ConflictException	409	Resource state conflict (e.g., duplicate entry).
UnprocessableEntityException	422	Business/semantic validation failed (entity well-formed but not processable).

Built-in exception families have fixed HTTP statuses. If you subclass BadRequestException, the response status remains HTTP 400; overriding http_code on that subclass does not change the family status. Use code for business-specific error codes inside the response body.

---

4. Upstream Service Errors

Use translate_upstream_errors around upstream httpx calls to convert known upstream failures into standardized ZodiacCore exceptions. The decorated function should call response.raise_for_status() so non-2xx responses can be classified.

from zodiac_core.http import ZodiacClient, translate_upstream_errors


@translate_upstream_errors(service="identity_and_access")
async def get_permissions(client: ZodiacClient):
    response = await client.post("/api/auth/by_user_id", json={"user_id": "..."})
    response.raise_for_status()
    return response.json()

Classification:

• Upstream HTTP 400 or 422 becomes UpstreamRequestException.
• Other upstream HTTP status failures become UpstreamServiceException.
• Other httpx.RequestError failures become UpstreamServiceException.

Some upstream services always return HTTP 200 and put business failures in their response body. In that case, parse the upstream payload and raise the ZodiacCore upstream exception yourself:

from zodiac_core.exceptions import UpstreamRequestException, UpstreamServiceException
from zodiac_core.http import ZodiacClient, translate_upstream_errors


@translate_upstream_errors(service="payment_gateway")
async def create_payment(client: ZodiacClient):
    response = await client.post("/payments", json={"amount": 100})
    response.raise_for_status()

    payload = response.json()
    if payload["code"] == "INVALID_CARD":
        raise UpstreamRequestException(
            service="payment_gateway",
            upstream_status=response.status_code,
        )

    if payload["code"] != "SUCCESS":
        raise UpstreamServiceException(
            service="payment_gateway",
            error_code="UPSTREAM_BUSINESS_ERROR",
            message="Upstream business error",
            upstream_status=response.status_code,
        )

    return payload["data"]

This keeps both integration styles on the same error path: standard RESTful httpx failures are translated by the decorator, while protocol-specific business codes can be converted explicitly by your application code.

These upstream exceptions are part of ZodiacCore's built-in exception hierarchy and standard exception registration:

from fastapi import FastAPI
from zodiac_core.exception_handlers import register_exception_handlers

app = FastAPI()
register_exception_handlers(app)

With the standard handlers registered, unhandled upstream exceptions return HTTP 400 and are not treated as uncaught HTTP 500 errors from the current service. If your service catches UpstreamServiceException or UpstreamRequestException itself, that local handling wins and the global handler is not involved.

---

5. Custom Exceptions

For a business error that belongs to a built-in HTTP status family, subclass the built-in exception and set a business code, message, and optional data:

from zodiac_core.exceptions import BadRequestException

class InsufficientBalanceException(BadRequestException):
    def __init__(self, current_balance: float):
        super().__init__(
            code=1001,  # Business error code; HTTP status stays 400
            message="Your account balance is too low.",
            data={"current_balance": current_balance},
        )

For a custom HTTP status that is not covered by the built-in exception families, inherit directly from ZodiacException and define http_code:

from fastapi import status
from zodiac_core.exceptions import ZodiacException

class RateLimitedException(ZodiacException):
    http_code = status.HTTP_429_TOO_MANY_REQUESTS

    def __init__(self, retry_after: int):
        super().__init__(
            code=2001,
            message="Too many requests.",
            data={"retry_after": retry_after},
        )

Usage in a route:

@app.post("/transfer")
async def transfer_money(amount: float):
    if amount > user.balance:
        raise InsufficientBalanceException(user.balance)
    ...

If a direct ZodiacException subclass does not define http_code, it inherits the default HTTP 500 status.

---

6. Integration

To enable global exception handling in your FastAPI app, use register_exception_handlers. This will catch:

1. All ZodiacException subclasses.
2. Pydantic ValidationError and FastAPI RequestValidationError (mapped to 422).
3. Upstream errors produced by translate_upstream_errors (mapped to 400).
4. Any uncaught Exception (mapped to 500 with secure logging).

from fastapi import FastAPI
from zodiac_core.exception_handlers import register_exception_handlers

app = FastAPI()
register_exception_handlers(app)

---

7. API Reference

Exception Base & Subclasses

ZodiacException

Bases: Exception

Base class for all zodiac-core related errors.

Source code in zodiac_core/exceptions.py

class ZodiacException(Exception):
    """Base class for all zodiac-core related errors."""

    http_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR

    def __init__(
        self,
        code: Optional[int] = None,
        data: Any = None,
        message: Optional[str] = None,
    ):
        self.code = code or self.http_code
        self.data = data
        if message is not None:
            self.message = message
        super().__init__(message)

BadRequestException

Bases: ZodiacException

Exception raised for 400 Bad Request errors.

Source code in zodiac_core/exceptions.py

class BadRequestException(ZodiacException):
    """Exception raised for 400 Bad Request errors."""

    http_code = status.HTTP_400_BAD_REQUEST

UnauthorizedException

Bases: ZodiacException

Exception raised for 401 Unauthorized errors.

Source code in zodiac_core/exceptions.py

class UnauthorizedException(ZodiacException):
    """Exception raised for 401 Unauthorized errors."""

    http_code = status.HTTP_401_UNAUTHORIZED

ForbiddenException

Bases: ZodiacException

Exception raised for 403 Forbidden errors.

Source code in zodiac_core/exceptions.py

class ForbiddenException(ZodiacException):
    """Exception raised for 403 Forbidden errors."""

    http_code = status.HTTP_403_FORBIDDEN

NotFoundException

Bases: ZodiacException

Exception raised for 404 Not Found errors.

Source code in zodiac_core/exceptions.py

class NotFoundException(ZodiacException):
    """Exception raised for 404 Not Found errors."""

    http_code = status.HTTP_404_NOT_FOUND

ConflictException

Bases: ZodiacException

Exception raised for 409 Conflict errors.

Source code in zodiac_core/exceptions.py

class ConflictException(ZodiacException):
    """Exception raised for 409 Conflict errors."""

    http_code = status.HTTP_409_CONFLICT

UnprocessableEntityException

Bases: ZodiacException

Exception raised for 422 Unprocessable Entity (business validation / semantic errors).

Source code in zodiac_core/exceptions.py

class UnprocessableEntityException(ZodiacException):
    """Exception raised for 422 Unprocessable Entity (business validation / semantic errors)."""

    http_code = status.HTTP_422_UNPROCESSABLE_CONTENT

UpstreamServiceException

Bases: BadRequestException

Exception raised when an upstream service is unavailable or fails unexpectedly.

Source code in zodiac_core/exceptions.py

class UpstreamServiceException(BadRequestException):
    """Exception raised when an upstream service is unavailable or fails unexpectedly."""

    def __init__(
        self,
        *,
        service: str,
        error_code: str = "UPSTREAM_SERVICE_ERROR",
        message: str = "Upstream service unavailable",
        upstream_status: Optional[int] = None,
    ):
        self.service = service
        self.error_code = error_code
        self.upstream_status = upstream_status
        super().__init__(
            message=message,
            data={
                "service": service,
                "error_code": error_code,
            },
        )

UpstreamRequestException

Bases: UpstreamServiceException

Exception raised when an upstream service rejects this service's request.

Source code in zodiac_core/exceptions.py

class UpstreamRequestException(UpstreamServiceException):
    """Exception raised when an upstream service rejects this service's request."""

    def __init__(
        self,
        *,
        service: str,
        upstream_status: Optional[int] = None,
    ):
        super().__init__(
            service=service,
            error_code="UPSTREAM_REQUEST_ERROR",
            message="Upstream request failed",
            upstream_status=upstream_status,
        )

Global Handler Registration

register_exception_handlers(app)

Register all exception handlers to the FastAPI app.

Order matters: 1. Specific Validation Errors 2. Specific Upstream Service Errors 3. Custom Business Logic Errors (ZodiacException) 4. Global Catch-All (Exception)

Source code in zodiac_core/exception_handlers.py

def register_exception_handlers(app: FastAPI) -> None:
    """
    Register all exception handlers to the FastAPI app.

    Order matters:
    1. Specific Validation Errors
    2. Specific Upstream Service Errors
    3. Custom Business Logic Errors (ZodiacException)
    4. Global Catch-All (Exception)
    """
    app.add_exception_handler(RequestValidationError, handler_validation_exception)
    app.add_exception_handler(ValidationError, handler_validation_exception)
    app.add_exception_handler(UpstreamServiceException, handler_upstream_service_exception)
    app.add_exception_handler(ZodiacException, handler_zodiac_exception)
    app.add_exception_handler(Exception, handler_global_exception)

Upstream HTTP Decorators

translate_upstream_errors(service)

Convert httpx upstream failures into standardized ZodiacCore exceptions.

Decorated functions should call response.raise_for_status() after receiving an httpx.Response so HTTP status failures can be classified.

Source code in zodiac_core/http.py

def translate_upstream_errors(service: str) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """
    Convert httpx upstream failures into standardized ZodiacCore exceptions.

    Decorated functions should call ``response.raise_for_status()`` after
    receiving an ``httpx.Response`` so HTTP status failures can be classified.
    """

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        if iscoroutinefunction(func):

            @wraps(func)
            async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> Any:
                try:
                    return await func(*args, **kwargs)
                except (httpx.HTTPStatusError, httpx.RequestError) as exc:
                    _raise_upstream_error(service, exc)

            return async_wrapper

        @wraps(func)
        def sync_wrapper(*args: P.args, **kwargs: P.kwargs) -> Any:
            try:
                return func(*args, **kwargs)
            except (httpx.HTTPStatusError, httpx.RequestError) as exc:
                _raise_upstream_error(service, exc)

        return sync_wrapper

    return decorator