Class Switchyard

switchyard_rust/core.py:465–625

Python-only compatibility chain for the current FastAPI/recipe surface.

class Switchyard:
    """Python-only compatibility chain for the current FastAPI/recipe surface."""

    state_key: ClassVar[str] = "switchyard"

    def __init__(
        self,
        *,
        request_processors: Iterable[Any] | None = None,
        backend: Any,
        response_processors: Iterable[Any] | None = None,
        translator: Any,
        fallback_target_on_evict: str | None = None,
    ) -> None:
        if not isinstance(backend, LLMBackend):
            actual = _role_type_name(backend)
            raise TypeError(f"Switchyard backend must be LLMBackend, got {actual}")
        self._request_components = tuple(request_processors or ())
        self._backend = backend
        self._response_components = tuple(response_processors or ())
        self._translator = translator
        self._fallback_target_on_evict = fallback_target_on_evict

    def iter_components(self) -> list[Any]:
        """Return lifecycle components in startup order."""
        return [
            *self._request_components,
            self._backend,
            *self._response_components,
            self._translator,
        ]

    async def call(
        self,
        request: Any,
        *,
        ctx: Any | None = None,
    ) -> Any:
        """Run the compatibility chain and translate the final response."""
        context = ctx if ctx is not None else ProxyContext()
        processed_request = await self._process_request_components(context, request)
        try:
            response = await self._call_backend_stage(context, processed_request)
        except _load_native().SwitchyardContextWindowExceededError as error:
            if self._fallback_target_on_evict is None:
                raise
            response = await self._retry_after_context_overflow(
                context,
                processed_request,
                error,
            )
        return await self._translator.translate(context, processed_request, response)

    async def _call_backend_stage(
        self,
        ctx: Any,
        request: Any,
    ) -> Any: