Skip to content

Routing & Response Wrapping

ZodiacCore enhances FastAPI's routing system to provide automatic response standardization. By using APIRouter, you ensure that every endpoint returns a consistent JSON structure without manual boilerplate.

1. The Zodiac APIRouter

The APIRouter in ZodiacCore is a drop-in replacement for fastapi.APIRouter. It uses a custom ZodiacRoute class that intercepts outgoing data and wraps it.

Automatic Wrapping

When you return a dictionary, a Pydantic model, or a list from your route, Zodiac automatically wraps it in a Response model:

from zodiac_core.routing import APIRouter

router = APIRouter()

@router.get("/status")
async def get_status():
    return {"status": "online"}

Resulting JSON:

{
  "code": 0,
  "message": "Success",
  "data": {
    "status": "online"
  }
}


2. Standard Response Structure

All Zodiac responses follow this schema:

Field Type Description
code int Business status code (0 for success).
message string A brief description of the result.
data any The actual payload (result of your function).

Manual Responses

If you need to return a non-standard response (e.g., a FileResponse or a custom status code), you can still return raw FastAPI Response objects or Zodiac's Response class. If the return type is already a Response, Zodiac will not wrap it again.

from zodiac_core.response import response_ok

@router.get("/custom")
async def manual():
    return response_ok(message="Custom success", data={"id": 1})

3. OpenAPI Integration

ZodiacCore's APIRouter dynamically generates Pydantic models for your responses. This means your Swagger UI (/docs) will correctly display the wrapped structure, including the code, message, and data fields, mapped to your specific return type.


4. API Reference

Routing Utilities

APIRouter

Bases: APIRouter

Zodiac-enhanced APIRouter that uses ZodiacRoute by default.

All routes registered via this router will automatically: - Wrap response_model with Response[T] for OpenAPI docs - Wrap endpoint return values with Response structure

Source code in zodiac_core/routing.py
class APIRouter(FastAPIRouter):
    """
    Zodiac-enhanced APIRouter that uses ZodiacRoute by default.

    All routes registered via this router will automatically:
    - Wrap response_model with Response[T] for OpenAPI docs
    - Wrap endpoint return values with Response structure
    """

    def __init__(self, *args, **kwargs):
        kwargs.setdefault("route_class", ZodiacRoute)
        super().__init__(*args, **kwargs)

ZodiacRoute

Bases: APIRoute

Custom APIRoute that automatically wraps response models and endpoint returns with the standard Response[T] structure.

Source code in zodiac_core/routing.py
class ZodiacRoute(APIRoute):
    """
    Custom APIRoute that automatically wraps response models and endpoint returns
    with the standard Response[T] structure.
    """

    def __init__(
        self,
        path: str,
        endpoint: Callable[..., Any],
        *,
        response_model: Any = None,
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        **kwargs,
    ) -> None:
        # Resolve FastAPI's DefaultPlaceholder
        if isinstance(response_model, DefaultPlaceholder):
            response_model = response_model.value

        # 1. Wrap main response model (default to Any if missing)
        if self._should_wrap(response_model):
            response_model = self._wrap_response_model(response_model or Any)

        # 2. Wrap additional responses (e.g. 400, 404 models)
        # Copy to avoid mutating caller's dict
        if responses:
            responses = {code: {**res_dict} for code, res_dict in responses.items()}
            for res in responses.values():
                if "model" in res and self._should_wrap(res["model"]):
                    res["model"] = self._wrap_response_model(res["model"])

        # 3. Wrap endpoint to auto-wrap return values
        endpoint = self._wrap_endpoint(endpoint)

        super().__init__(
            path,
            endpoint,
            response_model=response_model,
            responses=responses,
            **kwargs,
        )

    @staticmethod
    def _should_wrap(model: Any) -> bool:
        """Check if a model needs to be wrapped with Response[T]."""
        if model is Any or model is None:
            return True
        origin = get_origin(model)
        if origin is Response:
            return False
        try:
            if isinstance(model, type) and issubclass(model, Response):
                return False
        except TypeError:
            pass
        return True

    @staticmethod
    def _wrap_response_model(model: Any) -> type[Response]:
        """Wrap a model type with Response[T] using Pydantic's native generics."""
        return Response[model]

    @staticmethod
    def _maybe_wrap_result(result: Any) -> Any:
        """Wrap result in Response if not already a Response type."""
        if isinstance(result, (Response, FastAPIResponse)):
            return result
        return Response(data=result)

    @staticmethod
    def _wrap_endpoint(endpoint: Callable) -> Callable:
        """Wrap endpoint to automatically wrap return values in Response."""

        @wraps(endpoint)
        async def async_wrapper(*args, **kwargs):
            result = await endpoint(*args, **kwargs)
            return ZodiacRoute._maybe_wrap_result(result)

        @wraps(endpoint)
        def sync_wrapper(*args, **kwargs):
            result = endpoint(*args, **kwargs)
            return ZodiacRoute._maybe_wrap_result(result)

        return async_wrapper if inspect.iscoroutinefunction(endpoint) else sync_wrapper

Response Helpers

Response

Bases: BaseModel, Generic[T]

Standard API response model.

Source code in zodiac_core/response.py
class Response(BaseModel, Generic[T]):
    """Standard API response model."""

    model_config = ConfigDict(populate_by_name=True)

    code: int = Field(default=0, description="Business status code")
    data: Optional[T] = Field(default=None, description="Response payload")
    message: str = Field(default="Success", description="Response message")

create_response(http_code, code=None, data=None, message='')

Create a standardized JSON response.

Parameters:

Name Type Description Default
http_code int

HTTP status code

required
code Optional[int]

Business status code (defaults to http_code if not provided)

None
data Any

Response payload

None
message str

Response message

''
Source code in zodiac_core/response.py
def create_response(
    http_code: int,
    code: Optional[int] = None,
    data: Any = None,
    message: str = "",
) -> JSONResponse:
    """
    Create a standardized JSON response.

    Args:
        http_code: HTTP status code
        code: Business status code (defaults to http_code if not provided)
        data: Response payload
        message: Response message
    """
    if code is None:
        code = http_code

    response = Response(code=code, data=data, message=message)
    return JSONResponse(
        status_code=http_code,
        content=response.model_dump(mode="json"),
    )

response_ok(code=None, data=None, message='Success')

Create a successful response (200 OK)

Source code in zodiac_core/response.py
def response_ok(
    code: Optional[int] = None,
    data: Any = None,
    message: str = "Success",
) -> JSONResponse:
    """Create a successful response (200 OK)"""
    return create_response(status.HTTP_200_OK, code=code if code is not None else 0, data=data, message=message)

response_created(code=None, data=None, message='Created')

Create a resource created response (201 Created)

Source code in zodiac_core/response.py
def response_created(
    code: Optional[int] = None,
    data: Any = None,
    message: str = "Created",
) -> JSONResponse:
    """Create a resource created response (201 Created)"""
    return create_response(status.HTTP_201_CREATED, code=code, data=data, message=message)

response_bad_request(code=None, data=None, message='Bad Request')

Create a bad request error response (400 Bad Request)

Source code in zodiac_core/response.py
def response_bad_request(
    code: Optional[int] = None,
    data: Any = None,
    message: str = "Bad Request",
) -> JSONResponse:
    """Create a bad request error response (400 Bad Request)"""
    return create_response(status.HTTP_400_BAD_REQUEST, code=code, data=data, message=message)

response_unauthorized(code=None, data=None, message='Unauthorized')

Create an unauthorized response (401 Unauthorized)

Source code in zodiac_core/response.py
def response_unauthorized(
    code: Optional[int] = None,
    data: Any = None,
    message: str = "Unauthorized",
) -> JSONResponse:
    """Create an unauthorized response (401 Unauthorized)"""
    return create_response(status.HTTP_401_UNAUTHORIZED, code=code, data=data, message=message)

response_forbidden(code=None, data=None, message='Forbidden')

Create a forbidden response (403 Forbidden)

Source code in zodiac_core/response.py
def response_forbidden(
    code: Optional[int] = None,
    data: Any = None,
    message: str = "Forbidden",
) -> JSONResponse:
    """Create a forbidden response (403 Forbidden)"""
    return create_response(status.HTTP_403_FORBIDDEN, code=code, data=data, message=message)

response_not_found(code=None, data=None, message='Not Found')

Create a not found response (404 Not Found)

Source code in zodiac_core/response.py
def response_not_found(
    code: Optional[int] = None,
    data: Any = None,
    message: str = "Not Found",
) -> JSONResponse:
    """Create a not found response (404 Not Found)"""
    return create_response(status.HTTP_404_NOT_FOUND, code=code, data=data, message=message)

response_conflict(code=None, data=None, message='Conflict')

Create a conflict response (409 Conflict)

Source code in zodiac_core/response.py
def response_conflict(
    code: Optional[int] = None,
    data: Any = None,
    message: str = "Conflict",
) -> JSONResponse:
    """Create a conflict response (409 Conflict)"""
    return create_response(status.HTTP_409_CONFLICT, code=code, data=data, message=message)

response_unprocessable_entity(code=None, data=None, message='Unprocessable Entity')

Create an unprocessable entity response (422 Unprocessable Entity)

Source code in zodiac_core/response.py
def response_unprocessable_entity(
    code: Optional[int] = None,
    data: Any = None,
    message: str = "Unprocessable Entity",
) -> JSONResponse:
    """Create an unprocessable entity response (422 Unprocessable Entity)"""
    return create_response(status.HTTP_422_UNPROCESSABLE_CONTENT, code=code, data=data, message=message)

response_server_error(code=None, data=None, message='Internal Server Error')

Create a server error response (500 Internal Server Error)

Source code in zodiac_core/response.py
def response_server_error(
    code: Optional[int] = None,
    data: Any = None,
    message: str = "Internal Server Error",
) -> JSONResponse:
    """Create a server error response (500 Internal Server Error)"""
    return create_response(status.HTTP_500_INTERNAL_SERVER_ERROR, code=code, data=data, message=message)