Recipes
Each recipe is a single runnable script that solves one real integration problem end to end. Read them here, or download the whole kit:
Download the integration kit (.tar.gz) — every file below plus the docs, bundled.
recipes/01_health_probe.py
The minimum check before any other work: API reachable, key valid, per-task model lists loadable. Exits 0 on green.
recipes/01_health_probe.py
"""Recipe 01 — health probe.
The minimum check before any other work: is the API reachable, is your
key valid, are the per-task model lists loadable. Exits 0 on green, 1
on any failure.
GI_API_KEY=gi_… python3 01_health_probe.py
"""
from __future__ import annotations
import os
import sys
import requests
BASE_URL = os.environ.get("GI_BASE_URL", "https://api.genomicintelligence.ai")
API_KEY = os.environ.get("GI_API_KEY")
def fail(msg: str) -> None:
print(f"FAIL: {msg}", file=sys.stderr)
sys.exit(1)
def main() -> int:
if not API_KEY:
fail("set GI_API_KEY")
# 1. Public liveness — no auth required.
r = requests.get(f"{BASE_URL}/health", timeout=10)
if not r.ok:
fail(f"/health returned {r.status_code}: {r.text[:200]}")
health = r.json()
print(f"health : {health.get('status')} {health.get('version')}")
# 2. Authenticated round-trip — proves the bearer key is registered.
r = requests.get(
f"{BASE_URL}/v1/tasks/promoter/models",
headers={"Authorization": f"Bearer {API_KEY}"},
timeout=10,
)
if r.status_code == 401:
fail("401 unauthorized — key missing, malformed, or unrecognised. "
f"Check Authorization header. Body: {r.text[:200]}")
if not r.ok:
body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
err = body.get("error") or {}
fail(f"{r.status_code} {err.get('code', 'unknown')} request_id={err.get('request_id')}")
# Note: model-listing endpoints (`/v1/tasks/{task}/models`,
# `/v1/tasks/{task}/models/{id}/status`) return flat objects, NOT the
# `{data, meta}` envelope used by predict endpoints. Live OpenAPI at
# /v1/openapi.json is the source of truth for shapes per route.
body = r.json()
models = body.get("models", [])
print(f"models : {len(models)} registered for task=promoter "
f"(default={body.get('default_model')})")
# 3. Read RateLimit headers — confirms your tier and your headroom.
print(f"ratelimit: limit={r.headers.get('RateLimit-Limit')} "
f"remaining={r.headers.get('RateLimit-Remaining')} "
f"reset={r.headers.get('RateLimit-Reset')}s "
f"policy={r.headers.get('RateLimit-Policy')}")
print("ok")
return 0
if __name__ == "__main__":
raise SystemExit(main())
recipes/02_promoters_for_gene_list.py
Realistic batch shape: fetch each gene from Ensembl (public, no auth) and run promoter prediction, one BED-style row per detected region.
recipes/02_promoters_for_gene_list.py
"""Recipe 02 — promoter prediction across a list of human gene symbols.
Realistic batch shape: you have N gene symbols, you want promoters for
each. We fetch the genomic sequence for each gene from Ensembl REST
(public, no auth) and run `POST /v1/tasks/promoter/predict` against
each. Output is one BED-style row per detected promoter region.
Demonstrates:
- sequence acquisition from a public bioinformatics source (Ensembl)
- serial pacing against a per-key concurrency cap
- extraction of the typed `data.regions` field per the contract
GI_API_KEY=gi_… python3 02_promoters_for_gene_list.py TP53 MYC GAPDH
"""
from __future__ import annotations
import os
import sys
import time
from typing import Iterable
import requests
BASE_URL = os.environ.get("GI_BASE_URL", "https://api.genomicintelligence.ai")
API_KEY = os.environ.get("GI_API_KEY")
ENSEMBL = "https://rest.ensembl.org"
# Default genes to demo if the user gives no args. Sized for sub-2s per call.
DEFAULT_GENES = ["TP53", "MYC", "GAPDH"]
def fetch_gene_sequence(symbol: str, species: str = "human") -> tuple[str, str]:
"""Resolve `symbol` to an Ensembl ID and fetch its genomic sequence.
Returns (display_name, sequence). Raises on any HTTP failure.
Fetches on the gene's native strand so that minus-strand genes
(e.g. TP53) are returned in 5'→3' coding orientation — the
orientation the promoter model was trained on.
"""
# Lookup Ensembl ID + strand
r = requests.get(
f"{ENSEMBL}/lookup/symbol/{species}/{symbol}",
headers={"Accept": "application/json"},
timeout=15,
)
r.raise_for_status()
info = r.json()
chrom, start, end = info["seq_region_name"], info["start"], info["end"]
strand = info.get("strand", 1)
# Fetch genomic sequence on the gene's strand
region = f"{chrom}:{start}..{end}:{strand}"
r = requests.get(
f"{ENSEMBL}/sequence/region/{species}/{region}",
headers={"Accept": "text/plain"},
timeout=30,
)
r.raise_for_status()
return f"{symbol}|chr{chrom}:{start}-{end}|strand:{strand}", r.text.strip().upper()
def predict_promoter(sequence: str, sequence_name: str) -> tuple[dict, dict]:
"""One sync call to /v1/tasks/promoter/predict.
Returns ``(body, headers)`` — the parsed ``{data, meta}`` body plus the
response headers, so the caller can pace off the ``RateLimit-*`` spine.
"""
r = requests.post(
f"{BASE_URL}/v1/tasks/promoter/predict",
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
},
json={"sequence": sequence, "sequence_name": sequence_name,
"options": {"threshold": 0.5}},
timeout=120,
)
if not r.ok:
body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
err = body.get("error") or {}
raise RuntimeError(
f"{r.status_code} {err.get('code', 'unknown')}: "
f"{err.get('message', r.text[:200])} (request_id={err.get('request_id')})"
)
return r.json(), r.headers
def pace(headers) -> float:
"""Return how many seconds to sleep before the next call to stay at ~80% of capacity.
Reads IETF RateLimit-* headers. If headroom is low, slow down; if
plenty, no wait. Conservative pacing — beats reactive 429 retries.
"""
try:
remaining = int(headers.get("RateLimit-Remaining", "999"))
limit = int(headers.get("RateLimit-Limit", "10"))
reset = float(headers.get("RateLimit-Reset", "0"))
except (TypeError, ValueError):
return 0.0
threshold = max(1, int(limit * 0.2)) # aim to stay above 20% headroom
if remaining <= threshold and reset > 0:
# Spread remaining work across the reset window
return reset / max(1, remaining)
return 0.0
def run(symbols: Iterable[str]) -> int:
if not API_KEY:
print("ERROR: set GI_API_KEY", file=sys.stderr)
return 2
print("# track start end name score")
last_headers = {}
for symbol in symbols:
wait = pace(last_headers)
if wait > 0:
print(f"# pacing: sleep {wait:.2f}s", file=sys.stderr)
time.sleep(wait)
try:
display, sequence = fetch_gene_sequence(symbol)
except requests.HTTPError as exc:
print(f"# {symbol}: ensembl lookup failed — {exc}", file=sys.stderr)
continue
try:
body, last_headers = predict_promoter(sequence, display)
except RuntimeError as exc:
print(f"# {symbol}: predict failed — {exc}", file=sys.stderr)
continue
# `last_headers` now carries this call's RateLimit-* state, so the
# pace() at the top of the next iteration slows down before we run low.
regions = body.get("data", {}).get("regions") or []
for r in regions:
print(f"{symbol}\t{r['start']}\t{r['end']}\t{r.get('name', '.')}\t{r['score']:.4f}")
if not regions:
print(f"# {symbol}: 0 regions above threshold", file=sys.stderr)
return 0
if __name__ == "__main__":
args = sys.argv[1:] or DEFAULT_GENES
raise SystemExit(run(args))
recipes/03_async_annotation_polling.py
Async dispatch with progress polling for large inputs — HTTP status discriminates progress (202) from terminal (200/4xx/5xx).
recipes/03_async_annotation_polling.py
"""Recipe 03 — async annotation with progress polling.
Annotation is the slowest atomic task. For inputs above ~30 kbp, opt
into async to avoid the 300 s upstream proxy timeout. The polling shape
is uniform: HTTP status discriminates progress (`202`) vs terminal
(`200` success, `4xx`/`5xx` failure). Body is always `{data, meta}`
while running and on success; only terminal failure switches to the
error envelope.
Demonstrates:
- `Prefer: respond-async` opt-in
- 202-progress polling shape
- exponential backoff on the poll interval (don't hammer)
- terminal-state discrimination
GI_API_KEY=gi_… python3 03_async_annotation_polling.py path/to/sequence.fa
"""
from __future__ import annotations
import os
import sys
import time
import requests
BASE_URL = os.environ.get("GI_BASE_URL", "https://api.genomicintelligence.ai")
API_KEY = os.environ.get("GI_API_KEY")
HEADERS = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
def load_fasta(path: str) -> tuple[str, str]:
text = open(path).read().splitlines()
name = text[0].lstrip(">").strip()
sequence = "".join(line.strip() for line in text[1:] if line.strip()).upper()
return name, sequence
def submit_async(sequence: str, sequence_name: str) -> str:
"""POST with `Prefer: respond-async`; return the job_id."""
r = requests.post(
f"{BASE_URL}/v1/tasks/annotation/predict",
headers={**HEADERS, "Prefer": "respond-async"},
json={"sequence": sequence, "sequence_name": sequence_name,
"options": {"batch_size": 8}},
timeout=30,
)
r.raise_for_status()
return r.json()["data"]["job_id"]
def poll_until_terminal(job_id: str, max_wait_s: float = 30 * 60) -> dict:
"""Poll `/v1/tasks/jobs/{id}` until 200 or terminal 4xx/5xx.
Returns the final body. Raises on terminal failure or timeout.
"""
deadline = time.monotonic() + max_wait_s
interval = 2.0 # seconds; doubled up to a cap on each 202
while True:
r = requests.get(
f"{BASE_URL}/v1/tasks/jobs/{job_id}",
headers={"Authorization": f"Bearer {API_KEY}"},
timeout=30,
)
if r.status_code == 200:
return r.json() # done — {data, meta}
if r.status_code == 202:
# In flight. Body is also {data, meta}; data.progress carries
# current_percent / message / elapsed_seconds.
body = r.json()
progress = (body.get("data") or {}).get("progress") or {}
pct = progress.get("current_percent")
msg = progress.get("message", "")
print(f" [{pct:>3}%] {msg}", file=sys.stderr)
if time.monotonic() > deadline:
raise TimeoutError(f"job {job_id} did not terminate within {max_wait_s}s")
time.sleep(interval)
interval = min(interval * 1.5, 15.0) # back off, cap at 15s
continue
# Terminal failure — unified error envelope.
body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
err = body.get("error") or {}
raise RuntimeError(
f"job {job_id} terminal: {r.status_code} {err.get('code', 'unknown')} "
f"{err.get('message', '')} (request_id={err.get('request_id')})"
)
def main() -> int:
if not API_KEY:
print("ERROR: set GI_API_KEY", file=sys.stderr); return 2
if len(sys.argv) < 2:
print("usage: 03_async_annotation_polling.py <fasta-path>", file=sys.stderr); return 2
name, sequence = load_fasta(sys.argv[1])
print(f"submitting {len(sequence):,} bp as async annotation job…", file=sys.stderr)
job_id = submit_async(sequence, name)
print(f"job_id={job_id}", file=sys.stderr)
body = poll_until_terminal(job_id)
transcripts = body.get("data", {}).get("transcripts") or []
meta = body.get("meta", {})
print(f"done — transcripts={len(transcripts)} "
f"inference_ms={meta.get('inference_time_ms')} "
f"counts={meta.get('task_specific_counts')}")
for t in transcripts:
print(f" {t.get('name', '.')}\t{t.get('start')}\t{t.get('end')}\t"
f"{t.get('strand')}\t{t.get('score'):.4f}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
recipes/04_ratelimit_aware_retry.py
A call() wrapper that paces itself off the IETF RateLimit-* headers, honours Retry-After on 429, and backs off on 5xx.
recipes/04_ratelimit_aware_retry.py
"""Recipe 04 — rate-limit-aware request wrapper.
A `call()` helper that:
- reads IETF `RateLimit-*` headers from every response and paces the
next call to stay at ~80% of capacity
- on `429`, honours `Retry-After`; on `5xx`, exponential backoff
capped at 30 s
- never retries permanent 4xx errors (catalogue:
https://docs.genomicintelligence.ai/reference/errors.md)
- returns the parsed `{data, meta}` body on success or raises a
typed exception on permanent failure
Use this as a drop-in for `requests.post(...)` when you need pacing
without managing it by hand.
GI_API_KEY=gi_… python3 04_ratelimit_aware_retry.py
"""
from __future__ import annotations
import os
import time
from dataclasses import dataclass, field
from typing import Any, Optional
import requests
BASE_URL = os.environ.get("GI_BASE_URL", "https://api.genomicintelligence.ai")
API_KEY = os.environ.get("GI_API_KEY")
@dataclass
class GIError(Exception):
status: int
code: str
message: str
request_id: Optional[str] = None
details: Any = None
def __str__(self) -> str:
return f"[{self.status} {self.code}] {self.message} (request_id={self.request_id})"
@dataclass
class RateLimitState:
"""Sticky pacing state between calls. Threadsafe? No — per-worker."""
limit: int = 0
remaining: int = 0
reset_seconds: float = 0.0
last_response_at: float = field(default_factory=time.monotonic)
PERMANENT_4XX = {
"bad_request", "unauthorized", "forbidden", "not_found", "model_not_found",
"validation_failed", "task_not_supported_by_model", "payload_too_large",
"sync_too_large", "unsupported_format", "conflict", "job_expired",
}
RETRYABLE = {
"too_many_requests", # 429: rate or concurrency cap
"model_loading", # 503: warmup race
"service_unavailable", # 503: startup/shutdown
}
def _wait_before_next(state: RateLimitState) -> float:
"""Pacing: stay above 20% headroom of the token bucket."""
if state.limit <= 0:
return 0.0
threshold = max(1, int(state.limit * 0.2))
if state.remaining > threshold:
return 0.0
elapsed = time.monotonic() - state.last_response_at
remaining_window = max(0.0, state.reset_seconds - elapsed)
if remaining_window <= 0 or state.remaining <= 0:
return 0.0
return remaining_window / max(1, state.remaining)
def _read_headers(state: RateLimitState, headers) -> None:
try:
state.limit = int(headers.get("RateLimit-Limit", state.limit))
state.remaining = int(headers.get("RateLimit-Remaining", state.remaining))
state.reset_seconds = float(headers.get("RateLimit-Reset", state.reset_seconds))
except (TypeError, ValueError):
pass
state.last_response_at = time.monotonic()
def call(
method: str,
path: str,
*,
json_body: Optional[dict] = None,
headers: Optional[dict] = None,
state: RateLimitState,
max_attempts: int = 5,
) -> dict:
"""Issue one logical request with retries on retryable errors."""
url = f"{BASE_URL}{path}"
base_headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
if headers:
base_headers.update(headers)
backoff = 1.0
for attempt in range(1, max_attempts + 1):
wait = _wait_before_next(state)
if wait > 0:
time.sleep(wait)
r = requests.request(method, url, json=json_body, headers=base_headers, timeout=120)
_read_headers(state, r.headers)
if r.ok:
return r.json()
body = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
err = (body or {}).get("error") or {}
code = err.get("code", "unknown")
if code in PERMANENT_4XX:
# Don't retry; surface to caller.
raise GIError(r.status_code, code, err.get("message", ""),
err.get("request_id"), err.get("details"))
if code in RETRYABLE and attempt < max_attempts:
retry_after = r.headers.get("Retry-After")
sleep = float(retry_after) if retry_after else min(backoff, 30.0)
time.sleep(sleep)
backoff *= 2
continue
# 5xx server bug — one retry then give up.
if r.status_code >= 500 and attempt < max_attempts:
time.sleep(min(backoff, 30.0))
backoff *= 2
continue
raise GIError(r.status_code, code, err.get("message", ""),
err.get("request_id"), err.get("details"))
raise GIError(0, "exhausted", f"max_attempts={max_attempts} reached", None, None)
def demo() -> int:
if not API_KEY:
print("ERROR: set GI_API_KEY"); return 2
state = RateLimitState()
# A burst of small calls — the wrapper paces automatically.
for i in range(15):
try:
body = call(
"POST", "/v1/tasks/promoter/predict",
json_body={"sequence": "ACGT" * 500, "sequence_name": f"demo-{i}",
"options": {"threshold": 0.5}},
state=state,
)
ms = body.get("meta", {}).get("inference_time_ms")
print(f"call {i:>2}: {ms} ms (limit={state.limit} remaining={state.remaining})")
except GIError as exc:
print(f"call {i:>2}: FAILED — {exc}")
return 0
if __name__ == "__main__":
raise SystemExit(demo())
recipes/05_typed_error_handling.py
Switch on error.code, never on error.message. Maps every error code to retryable / permanent / server-bug plus the action to take.
recipes/05_typed_error_handling.py
"""Recipe 05 — typed error handling.
Switch on `error.code`, never on `error.message`. Every code from
https://docs.genomicintelligence.ai/reference/errors.md is mapped here
to one of four buckets — retryable, permanent, server-bug, or
proxy-timeout (the enveloped-less 504 special case) — plus the action
your code should take.
Use this as the template for your own error handler. Copy `handle()`
into your codebase, register it as your one place to translate API
errors to your domain's exception types.
GI_API_KEY=gi_… python3 05_typed_error_handling.py
"""
from __future__ import annotations
import os
import sys
from dataclasses import dataclass
from typing import Any, Optional
import requests
BASE_URL = os.environ.get("GI_BASE_URL", "https://api.genomicintelligence.ai")
API_KEY = os.environ.get("GI_API_KEY")
@dataclass
class APIError(Exception):
status: int
code: str
message: str
request_id: Optional[str]
details: Any
bucket: str # "retryable" | "permanent" | "server_bug" | "proxy_timeout"
action: str # human-readable action hint for the caller
def handle(response: requests.Response) -> dict:
"""Convert an API response into either a parsed body or a raised APIError.
Single source of truth for every code in the catalogue. If we ever add
a new error code, the `unknown_code` branch surfaces it cleanly.
"""
if response.ok:
return response.json()
# 504 from the edge proxy is the one response on the contract that
# is NOT the unified error envelope. Treat as a special case.
if response.status_code == 504:
raise APIError(
status=504, code="gateway_timeout",
message="Upstream proxy read timeout (300 s).",
request_id=None, details=None,
bucket="proxy_timeout",
action="Retry the same request with header `Prefer: respond-async`.",
)
body = response.json() if response.headers.get("content-type", "").startswith("application/json") else {}
err = (body or {}).get("error") or {}
code = err.get("code", "unknown_code")
msg = err.get("message", "")
request_id = err.get("request_id")
details = err.get("details")
# Retryable transient — the API or upstream is asking you to back off.
if code == "too_many_requests":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="retryable",
action="Honour Retry-After. Both rate-bucket and concurrency 429s use this code.")
if code == "model_loading":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="retryable",
action="Wait ~30 s and retry. The model is warming up.")
if code == "service_unavailable":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="retryable",
action="Backoff and retry. Server is starting up or shutting down.")
# Permanent client errors — surface to the caller, don't retry.
if code == "bad_request":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="permanent",
action="Fix the HTTP-level precondition and re-issue.")
if code == "unauthorized":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="permanent",
action="Check Authorization header. WWW-Authenticate splits invalid_request vs invalid_token.")
if code == "forbidden":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="permanent",
action="Key is registered but disabled. Contact the API operator.")
if code == "not_found":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="permanent",
action="Unknown route, unknown job, or cross-tenant access (returned as 404 not 403).")
if code == "model_not_found":
# details.available_models is the typed list of alternatives.
raise APIError(response.status_code, code, msg, request_id, details,
bucket="permanent",
action=f"Pick a model from details.available_models: {details}.")
if code == "validation_failed":
# details.errors is the FastAPI per-field array.
raise APIError(response.status_code, code, msg, request_id, details,
bucket="permanent",
action="Surface message verbatim — it's deterministic. See details.errors per-field.")
if code == "task_not_supported_by_model":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="permanent",
action="Pick a different model — see details.supported_tasks.")
if code == "payload_too_large":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="permanent",
action="Body exceeded 16 MiB. Split the request or use sequence references if available.")
if code == "sync_too_large":
# Reserved in the schema; not emitted today but typed for the future.
raise APIError(response.status_code, code, msg, request_id, details,
bucket="permanent",
action="Switch to async (`Prefer: respond-async`).")
if code == "unsupported_format":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="permanent",
action="The Accept / ?format= value is not in this task's allowed set.")
if code == "conflict":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="permanent",
action="Reserved (e.g. busy-model unload conflict). Retry once after delay.")
if code == "job_expired":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="permanent",
action="Async result evicted (24 h TTL or process restart). Resubmit the original POST.")
# Server bug — capture and escalate.
if code == "internal_error":
raise APIError(response.status_code, code, msg, request_id, details,
bucket="server_bug",
action=f"Email contact@genomicintelligence.ai with request_id={request_id}.")
# Unknown code — log loudly so you notice when the catalogue grows.
raise APIError(response.status_code, code, msg, request_id, details,
bucket="server_bug",
action=f"Unknown error.code={code!r}. Treat as server bug; capture and escalate.")
def demo() -> int:
if not API_KEY:
print("ERROR: set GI_API_KEY"); return 2
# Trigger 422 validation_failed by sending a non-DNA character.
print("# trigger validation_failed (422)")
r = requests.post(
f"{BASE_URL}/v1/tasks/promoter/predict",
headers={"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"},
json={"sequence": "ZZZZ", "sequence_name": "bad-input"},
timeout=30,
)
try:
handle(r)
except APIError as exc:
print(f" caught: {exc.code} bucket={exc.bucket}")
print(f" action: {exc.action}")
print(f" message: {exc.message}")
# Trigger 401 unauthorized with a clearly-bad key.
print("\n# trigger unauthorized (401)")
r = requests.get(
f"{BASE_URL}/v1/tasks/promoter/models",
headers={"Authorization": "Bearer gi_definitely_not_a_real_key"},
timeout=10,
)
try:
handle(r)
except APIError as exc:
print(f" caught: {exc.code} bucket={exc.bucket}")
print(f" action: {exc.action}")
print(f" www-authenticate: {r.headers.get('WWW-Authenticate')}")
return 0
if __name__ == "__main__":
raise SystemExit(demo())