bioptimus.models.clients.http

HTTP client for direct API endpoints. Sends JSON payloads to a model server via plain HTTP POST using requests (synchronous) and aiohttp (asynchronous).

is_transient

def is_transient(exc: BaseException) -> bool

Returns whether an exception is a transient failure worth retrying. Transient failures include connection resets, timeouts, and HTTP responses carrying a retryable server-side status code (e.g. 503 from momentary GPU back-pressure). Deterministic failures such as HTTP 400 (malformed request) are not retried.

The exception raised while dispatching a request.

Returns: True if the failure is transient and should be retried.

tile_retry_logger

def tile_retry_logger(
    logger: logging.Logger,
    tile_index: int,
    *,
    max_retries: int = DEFAULT_MAX_RETRIES
) -> Callable[[int, BaseException, float], None]

Builds an on_retry callback that logs a per-tile dispatch retry. Returns the callback expected by retry_sync / retry_async, emitting a structured tile_dispatch_retry warning. Shared by every per-tile dispatch path so the retry log event and fields stay identical.

The logger to emit on.

Index of the tile being retried, recorded in the event.

Retry budget, used to report the total attempt count.

Returns: A callback taking (attempt, exc, delay) that logs the retry.

retry_sync

def retry_sync(
        attempt: Callable[[], T],
        *,
        max_retries: int = DEFAULT_MAX_RETRIES,
        base_delay: float = DEFAULT_RETRY_BASE_DELAY,
        is_retryable: Callable[[BaseException], bool] = is_transient,
        sleep: Callable[[float], None] = time.sleep,
        on_retry: Callable[[int, BaseException, float], None] | None = None
) -> T

Runs attempt, retrying transient failures with exponential backoff. Synchronous twin of retry_async for threaded (non-async) callers, such as the TorchModel thread-pool dispatch. Shares the same backoff policy, defaults, and on_retry contract so both paths behave identically.

Zero-argument callable that performs one attempt.

Number of retries after the initial attempt (so the total number of attempts is max_retries + 1).

Seed for the exponential backoff (base_delay * 2**n).

Predicate deciding whether a raised exception is transient and worth retrying.

Blocking sleep function; injectable so tests can skip real waits.

Optional callback invoked before each retry with the zero-based attempt number, the exception, and the delay.

Returns: The result of the first successful attempt. Raises:

BaseException - Re-raises the last exception when the retry budget is exhausted or the exception is not retryable.

retry_async

async def retry_async(
        attempt: Callable[[], Awaitable[T]],
        *,
        max_retries: int = DEFAULT_MAX_RETRIES,
        base_delay: float = DEFAULT_RETRY_BASE_DELAY,
        is_retryable: Callable[[BaseException], bool] = is_transient,
        sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
        on_retry: Callable[[int, BaseException, float], None] | None = None
) -> T

Runs attempt, retrying transient failures with exponential backoff. The retry mechanism is transport-agnostic: the work to perform, the logging, and (optionally) the sleep function are all supplied by the caller, keeping this function reusable and easy to test.

Zero-argument coroutine factory that performs one attempt.

Number of retries after the initial attempt (so the total number of attempts is max_retries + 1).

Seed for the exponential backoff (base_delay * 2**n).

Predicate deciding whether a raised exception is transient and worth retrying.

Async sleep function; injectable so tests can skip real waits.

Optional callback invoked before each retry with the zero-based attempt number, the exception, and the delay.

Returns: The result of the first successful attempt. Raises:

BaseException - Re-raises the last exception when the retry budget is exhausted or the exception is not retryable.

HTTPClient

class HTTPClient(base_url: str, endpoints: Dict[str, str], timeout: float = 30.0)

Sends JSON payloads over HTTP. Fully configured at construction — predict / embed only need the serialized body.

Root URL of the model server, e.g. "http://localhost:8080".

Mapping of endpoint type to path. At least one key must be present; see the example below.

HTTP request timeout in seconds.

Example:

client = HTTPClient(
    "http://localhost:8080",
    endpoints={
        "prediction": "/api/predict/m-optimus",
        "embedding": "/api/embed/m-optimus",
    },
)
pred_json = client.predict(request_json)
emb_json = client.embed(request_json)

close

def close() -> None

Closes the underlying HTTP session and releases connections.

predict

@log_failures
def predict(body: str) -> str

POST to the prediction endpoint.

Serialized JSON request payload.

Returns: Response body as a JSON string.

embed

@log_failures
def embed(body: str) -> str

POST to the embedding endpoint.

Serialized JSON request payload.

Returns: Response body as a JSON string.

predict_with_embedding

@log_failures
def predict_with_embedding(body: str) -> str

POST to the combined prediction+embedding endpoint. Returns both the gene prediction and the tile embedding from a single forward pass. Only models that expose both prediction and embedding configure this endpoint; others raise NotImplementedError.

Serialized JSON request payload.

Returns: Response body as a JSON string, with embedding populated alongside output.

metadata

@log_failures
def metadata() -> str

GET the metadata endpoint. Returns: Response body as a JSON string.

predict_async

@log_failures
async def predict_async(body: str, session: Any = None) -> str

POST to the prediction endpoint asynchronously.

Serialized JSON request payload.

An aiohttp.ClientSession for connection pooling.

Returns: Response body as a JSON string.

embed_async

@log_failures
async def embed_async(body: str, session: Any = None) -> str

POST to the embedding endpoint asynchronously.

Serialized JSON request payload.

An aiohttp.ClientSession for connection pooling.

Returns: Response body as a JSON string.

predict_with_embedding_async

@log_failures
async def predict_with_embedding_async(body: str, session: Any = None) -> str

POST to the combined prediction+embedding endpoint asynchronously.

Serialized JSON request payload.

An aiohttp.ClientSession for connection pooling.

Returns: Response body as a JSON string, with embedding populated alongside output.

Overview

Get Started

Preprocessing

Workflows

Troubleshooting

Reference

is_transient

tile_retry_logger

retry_sync

retry_async

HTTPClient

close

predict

embed

predict_with_embedding

metadata

predict_async

embed_async

predict_with_embedding_async

​is_transient

​tile_retry_logger

​retry_sync

​retry_async

​HTTPClient

​close

​predict

​embed

​predict_with_embedding

​metadata

​predict_async

​embed_async

​predict_with_embedding_async

is_transient

tile_retry_logger

retry_sync

retry_async

HTTPClient

close

predict

embed

predict_with_embedding

metadata

predict_async

embed_async

predict_with_embedding_async