API Reference

Gateway core

Transport-agnostic gateway core.

Gateway.handle takes a plain (method, path, headers, body) request and returns a Response. The HTTP server is a thin translation on top of this; streaming SSE lives in the server because it needs the raw socket.

class relational_transformers_gateway.gateway.Response(status: 'int', body: 'bytes', content_type: 'str' = 'application/json', headers: 'dict[str, str]'=<factory>)
Parameters:

  • status (int)

  • body (bytes)

  • content_type (str)

  • headers (dict[str, str])

status: int
body: bytes
content_type: str = 'application/json'
headers: dict[str, str]
classmethod error(status, message, headers=None)
Parameters:

  • status (int)

  • message (str)

  • headers (dict[str, str] | None)

Return type:

Response

relational_transformers_gateway.gateway.model_tokens(path, body)

Transformer work units for a worker request: b * s for /v1/forward.

Parameters:

  • path (str)

  • body (bytes)

Return type:

int

class relational_transformers_gateway.gateway.Gateway(config, ddb=None, discovery=None, scorer=None)
Parameters:

  • config (Config)

  • ddb (Any)

  • discovery (Any)

  • scorer (Any)

handle(method, path, headers, body)

Route one non-streaming request end to end.

Parameters:

  • method (str)

  • path (str)

  • headers (dict[str, str])

  • body (bytes)

Return type:

Response

Configuration

Environment-driven gateway configuration, validated once at startup.

class relational_transformers_gateway.config.Config(cloudmap_namespace, cloudmap_service, keys_table, usage_table, upstream_timeout=25.0, max_body_bytes=10485760, discovery_cache_seconds=5.0, auth_cache_seconds=30.0, worker_attempts=3, rate_limit_rps=0.0, rate_limit_burst=0.0, openai_api_key='', openai_base_url='https://api.openai.com', openai_allowed_models=<factory>, usage_retention_days=90, model_checkpoint='')

Every knob the gateway reads, resolved from the environment.

Parameters:

  • cloudmap_namespace (str)

  • cloudmap_service (str)

  • keys_table (str)

  • usage_table (str)

  • upstream_timeout (float)

  • max_body_bytes (int)

  • discovery_cache_seconds (float)

  • auth_cache_seconds (float)

  • worker_attempts (int)

  • rate_limit_rps (float)

  • rate_limit_burst (float)

  • openai_api_key (str)

  • openai_base_url (str)

  • openai_allowed_models (frozenset[str])

  • usage_retention_days (int)

  • model_checkpoint (str)

Authentication and rate limiting

API-key authentication against the DynamoDB keys table.

Keys are opaque bearer tokens; only their SHA-256 digests are stored. A small positive/negative cache bounds DynamoDB reads under load — set AUTH_CACHE_SECONDS=0 to look every request up fresh.

class relational_transformers_gateway.auth.RateLimiter(rps, burst)

Per-customer token bucket. rps <= 0 disables limiting.

Parameters:

  • rps (float)

  • burst (float)

Worker discovery

Worker discovery through AWS Cloud Map, with a short shared cache.

Metering

Atomic per-request usage metering in DynamoDB.

Each completed request writes an expiring ledger row and increments the customer’s monthly rollup in one transaction. The ledger row is conditioned on the request id being new, so a replayed request id cannot double-bill.

exception relational_transformers_gateway.metering.DuplicateRequestError

The request id was already metered.

class relational_transformers_gateway.metering.Usage(customer_id: 'str', request_id: 'str', path: 'str', status: 'int', data_in_bytes: 'int', data_out_bytes: 'int', model_tokens: 'int' = 0, input_tokens: 'int' = 0, output_tokens: 'int' = 0, latency_ms: 'int' = 0)
Parameters:

  • customer_id (str)

  • request_id (str)

  • path (str)

  • status (int)

  • data_in_bytes (int)

  • data_out_bytes (int)

  • model_tokens (int)

  • input_tokens (int)

  • output_tokens (int)

  • latency_ms (int)

LLM passthrough

OpenAI-compatible LLM passthrough: validation, usage, upstream requests.

relational_transformers_gateway.llm.prepare_body(config, path, body)

Validate an LLM request body; return (upstream body, is_streaming).

Streaming chat requests get stream_options.include_usage forced on so the final SSE chunk carries token counts for metering.

Parameters:

  • config (Config)

  • path (str)

  • body (bytes)

Return type:

tuple[bytes, bool]

relational_transformers_gateway.llm.extract_usage(payload)

(input, output, total) tokens from either OpenAI response shape.

Parameters:

payload (dict)

Return type:

tuple[int, int, int]

relational_transformers_gateway.llm.filter_models(config, body)

Restrict a /v1/models listing to the models this gateway serves.

Parameters:

  • config (Config)

  • body (bytes)

Return type:

bytes

HTTP server

Persistent HTTP front end with SSE streaming for chat completions.