core.requester

Requester object to make HTTP requests to the Jobmon FastAPI services.

Attributes

logger

Classes

Requester

Handles HTTP requests to the jobmon server with configurable OTLP integration.

Functions

http_request_ok(→ bool)	Return True if HTTP return codes that are deemed ok.
get_content(→ Tuple[int, Any])	Parse the response.

Module Contents

core.requester.logger

core.requester.http_request_ok(status_code: int) → bool

Return True if HTTP return codes that are deemed ok.

class core.requester.Requester(service_url: str, retries_timeout: int = 300, retries_attempts: int = 10, request_timeout: int = 20, use_otlp: bool = False)

Handles HTTP requests to the jobmon server with configurable OTLP integration.

_otlp_manager = None

service_url

retries_timeout = 300

retries_attempts = 10

request_timeout = 20

server_structlog_context: Dict[str, str]

classmethod _init_otlp() → None

Initialize OTLP tracing only - logging handled by client configuration.

classmethod from_defaults() → Requester

Instantiate a requester from default config values.

property url: str

Legacy property for backward compatibility.

add_server_structlog_context(**kwargs: Any) → None

Add the structlogging context if it has been provided.

_get_current_structlog_context() → Dict[str, Any]

Get current structlog context variables to pass to server.

Captures context like workflow_run_id, cluster_name, etc. so the server can include them in its logs for correlation.

tracing_span(app_route: str, request_type: str) → Any

_maybe_trace(func: Callable) → Callable

_should_retry_exception(exception: Any) → bool

Determine if an exception should trigger a retry.

_maybe_retry(func: Callable, tenacious: bool) → Any

_send_request(app_route: str, message: dict, request_type: str) → Tuple[int, Any]

send_request(app_route: str, message: dict, request_type: str, tenacious: bool = True) → Tuple[int, Any]

Send a request to the Jobmon server.

async _send_request_async(session: aiohttp.ClientSession, app_route: str, message: dict, request_type: str) → Tuple[int, Any]

Async version of _send_request using aiohttp.

async _get_content_async(response: aiohttp.ClientResponse) → Tuple[int, Any]

Parse an aiohttp response, handling JSON and non-JSON content gracefully.

Parameters:

response – The aiohttp ClientResponse object to parse.

Returns:

Tuple of (status_code, content) where content is parsed JSON or raw text/bytes.

_maybe_retry_async(func: Callable, tenacious: bool) → Any

Async version of _maybe_retry using tenacity async retry.

async send_request_async(session: aiohttp.ClientSession, app_route: str, message: dict, request_type: str, tenacious: bool = True) → Tuple[int, Any]

Send an async request to the Jobmon server with sophisticated retry logic.

This method provides the same robust retry capabilities as the sync version, using tenacity for exponential backoff with jitter, timeout protection, and comprehensive exception handling.

Parameters:

• session – An active aiohttp ClientSession for making requests.

• app_route – The API route to request (will be appended to base URL).

• message – Dictionary containing the request payload.

• request_type – HTTP method - ‘get’, ‘post’, or ‘put’.

• tenacious – Whether to enable retry logic (default: True).

Returns:

Tuple of (status_code, response_content).

Raises:

• InvalidRequest – For 4xx client errors (no retry).

• InvalidResponse – For 5xx server errors after exhausting retries.

• RuntimeError – If retry budget is exceeded.

core.requester.get_content(response: Any) → Tuple[int, Any]

Parse the response.