Routing & Response Wrapping

ZodiacCore enhances FastAPI's routing system with default response standardization. Body-bearing endpoints use a consistent JSON structure without manual boilerplate, while raw response objects and HTTP no-body response semantics remain available.

1. The Zodiac APIRouter

The APIRouter in ZodiacCore accepts FastAPI-style route declarations while adding a mandatory envelope for ordinary response data. It uses a custom ZodiacRoute class to wrap response models and returned values.

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"
  }
}

When response_model is omitted, Zodiac infers the payload type from the endpoint's return annotation. An endpoint without a return annotation uses Response[Any].

Response Model Semantics

Route declaration	Runtime behavior	OpenAPI response model
Omitted `response_model`, return type `T`	Automatically wrapped	`Response[T]`
Omitted `response_model`, no return type	Automatically wrapped	`Response[Any]`
`response_model=T`	Automatically wrapped	`Response[T]`
`response_model=None`, ordinary return value	Automatically wrapped without constraining `data`	`Response[Any]`
`response_model` omitted or set to `None`, return type is a FastAPI/Starlette `Response`	Raw response is passed through	No Zodiac envelope model
Status code `<200`, `204`, `205`, or `304`	No response body or envelope	No response content

Declaring a non-empty response model for a status code that prohibits a body remains an error, matching FastAPI.

Unlike FastAPI's native router, Zodiac intentionally treats response_model=None as an untyped payload rather than as an envelope opt-out. This preserves the code, data, and message contract. Runtime passthrough is based on the actual returned FastAPI/Starlette Response object; its return annotation keeps OpenAPI aligned with that raw response.

2. Standard Response Structure

Body-bearing responses wrapped by Zodiac 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

Raw FastAPI/Starlette Response objects, such as FileResponse and StreamingResponse, are passed through without runtime wrapping. Annotate the raw response return type so OpenAPI does not advertise a Zodiac envelope.

from fastapi import Response

@router.get("/custom")
async def manual() -> Response:
    return Response("custom", media_type="text/plain")

Yield-Based Streaming

When the installed FastAPI provides native yield-based streaming, Zodiac leaves synchronous and asynchronous generator endpoints unchanged. Streamed chunks are not wrapped in the Zodiac Response envelope.

FastAPI 0.134.0 added yield-based JSON Lines (JSONL) and binary streaming. A generator using the default response class produces JSONL, while response_class=StreamingResponse sends raw strings or bytes:

from collections.abc import AsyncIterable, Iterator

from fastapi.responses import StreamingResponse
from pydantic import BaseModel


class User(BaseModel):
    id: int
    name: str


@router.get("/users/stream")
async def stream_users() -> AsyncIterable[User]:
    yield User(id=1, name="First")
    yield User(id=2, name="Second")


@router.get("/download", response_class=StreamingResponse)
def stream_download() -> Iterator[bytes]:
    yield b"first chunk\n"
    yield b"second chunk\n"

FastAPI 0.135.0 added native Server-Sent Events (SSE). EventSourceResponse is a StreamingResponse subclass, so it receives the same passthrough behavior:

from collections.abc import AsyncIterable

from fastapi.sse import EventSourceResponse


@router.get("/events", response_class=EventSourceResponse)
async def stream_events() -> AsyncIterable[dict[str, str]]:
    yield {"event": "ready"}
    yield {"event": "complete"}

FastAPI versions before 0.134.0 do not provide native yield-based streaming, so Zodiac does not opt generator endpoints into passthrough on those versions. Unannotated generators retain the ordinary response-envelope behavior; iterator return annotations are still subject to the older FastAPI/Pydantic response-model limitations and may fail during route registration. For streaming that must also work on older FastAPI versions, explicitly return a StreamingResponse object instead. See FastAPI's guides for JSONL streaming, raw stream data, and SSE.

3. OpenAPI Integration

ZodiacCore's APIRouter dynamically generates Pydantic models for wrapped responses. Swagger UI (/docs) displays the code, message, and data fields, with data mapped to an explicit response model or the endpoint's return annotation. response_model=None produces Response[Any]; annotated raw responses and no-body status codes do not generate a Zodiac envelope schema.

4. API Reference

Routing Utilities

`APIRouter`

Bases: APIRouter

Zodiac-enhanced APIRouter that uses ZodiacRoute by default.

Body-bearing routes registered via this router will, by default: - Infer omitted response models from endpoint return annotations - Wrap response models with Response[T] for OpenAPI docs - Wrap endpoint return values with the Response structure

response_model=None keeps the envelope with an unconstrained Any payload. Return a FastAPI/Starlette Response object to bypass runtime wrapping. Generator endpoints and StreamingResponse classes retain FastAPI's native streaming behavior and are not wrapped in the Response structure.

Source code in zodiac_core/routing.py

class APIRouter(FastAPIRouter):
    """
    Zodiac-enhanced APIRouter that uses ZodiacRoute by default.

    Body-bearing routes registered via this router will, by default:
    - Infer omitted response models from endpoint return annotations
    - Wrap response models with Response[T] for OpenAPI docs
    - Wrap endpoint return values with the Response structure

    response_model=None keeps the envelope with an unconstrained Any payload.
    Return a FastAPI/Starlette Response object to bypass runtime wrapping.
    Generator endpoints and StreamingResponse classes retain FastAPI's native
    streaming behavior and are not wrapped in the Response structure.
    """

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

`ZodiacRoute`

Bases: APIRoute

Custom APIRoute that wraps body-bearing response models and endpoint returns with the standard Response[T] structure by default.

Raw Response objects pass through at runtime, while status codes that prohibit a response body bypass automatic wrapping entirely.

Source code in zodiac_core/routing.py

class ZodiacRoute(APIRoute):
    """
    Custom APIRoute that wraps body-bearing response models and endpoint returns
    with the standard Response[T] structure by default.

    Raw Response objects pass through at runtime, while status codes that
    prohibit a response body bypass automatic wrapping entirely.
    """

    def __init__(
        self,
        path: str,
        endpoint: Callable[..., Any],
        *,
        response_model: Any = _DEFAULT_RESPONSE_MODEL,
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        **kwargs,
    ) -> None:
        # Preserve FastAPI's omitted-model sentinel so return annotations can be
        # inferred separately from Zodiac's response_model=None -> Any contract.
        response_model_is_default = isinstance(response_model, DefaultPlaceholder)
        body_allowed = is_body_allowed_for_status_code(kwargs.get("status_code"))
        return_annotation = get_typed_return_annotation(endpoint)
        raw_return_annotation = _unwrap_annotated(return_annotation)
        returns_raw_response = lenient_issubclass(raw_return_annotation, FastAPIResponse)
        is_streaming_endpoint = self._is_streaming_endpoint(endpoint, kwargs.get("response_class"))

        if not is_streaming_endpoint:
            if response_model_is_default:
                if returns_raw_response:
                    response_model = None
                elif return_annotation is None:
                    response_model = Any if body_allowed else None
                else:
                    response_model = return_annotation
            elif response_model is None and body_allowed and not returns_raw_response:
                response_model = Any

        # 1. Wrap the inferred or explicit main response model.
        if not is_streaming_endpoint and self._should_wrap(response_model):
            response_model = self._wrap_response_model(response_model)

        # 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. Non-streaming body-bearing endpoints apply the runtime wrapper. The
        # result check passes actual FastAPI/Starlette Response objects through.
        if body_allowed and not is_streaming_endpoint:
            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]."""
        model = _unwrap_annotated(model)
        return model is not None and not lenient_issubclass(model, Response)

    @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 _callable_candidates(
        endpoint: Callable[..., Any],
    ) -> tuple[Callable[..., Any], Callable[..., Any]]:
        """Return callable variants before and after decorator unwrapping."""
        while isinstance(endpoint, partial):
            endpoint = endpoint.func
        return endpoint, inspect.unwrap(endpoint)

    @staticmethod
    def _is_generator_callable(endpoint: Callable[..., Any]) -> bool:
        """Return whether a function or callable object produces a generator."""

        def is_generator(candidate: Any) -> bool:
            return inspect.isgeneratorfunction(candidate) or inspect.isasyncgenfunction(candidate)

        candidates = ZodiacRoute._callable_candidates(endpoint)
        if any(is_generator(candidate) for candidate in candidates):
            return True
        if inspect.isclass(candidates[-1]):
            return False
        for candidate in candidates:
            if callable(candidate):
                calls = ZodiacRoute._callable_candidates(candidate.__call__)
                if any(is_generator(call) for call in calls):
                    return True
        return False

    @staticmethod
    def _is_streaming_endpoint(endpoint: Callable[..., Any], response_class: Any) -> bool:
        """Return whether FastAPI must retain control of the endpoint's stream."""
        if _SUPPORTS_YIELD_STREAMING and ZodiacRoute._is_generator_callable(endpoint):
            return True
        if isinstance(response_class, DefaultPlaceholder):
            response_class = response_class.value
        return lenient_issubclass(response_class, StreamingResponse)

    @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