#!/usr/bin/env python3 """Template client for the Phage Annotation Server web services. Submits a genome, waits for it, downloads the results, and parses them -- using only the Python standard library, so it runs anywhere with Python 3.9+ and nothing to install. python3 api_client.py my_phage.fasta python3 api_client.py --example # server-side sample genome python3 api_client.py my_phage.fasta --out ./results --server https://... Copy this file and adapt it; it is meant as a starting point, not a package. WHAT IT DEMONSTRATES 1. POST /jobs submit (multipart), asking for JSON 2. GET /jobs/{id}/status.json poll until a terminal state 3. GET /jobs/{id}/browser.json structured per-contig annotations 4. GET /jobs/{id}/files/{path} an individual result file 5. GET /jobs/{id}/download the complete results archive ... then PARSES the phynteny TSV and prints a per-category summary. NOTES * Send `Accept: application/json` on submission. Without it the server does a browser-style 303 redirect and you have to scrape the Location header for the job id. * Jobs are rate-limited per IP. Be polite in loops; the poll interval below is deliberately not aggressive. * Results are deleted after the server's retention window -- download anything you need to keep. """ from __future__ import annotations import argparse import csv import io import json import mimetypes import os import sys import time import urllib.error import urllib.request import uuid import zipfile from collections import Counter from pathlib import Path DEFAULT_SERVER = "https://phage-annotation.org" POLL_SECONDS = 10 POLL_TIMEOUT_SECONDS = 60 * 90 # --------------------------------------------------------------------------- # # Minimal multipart encoder (stdlib has no request-side multipart helper) # --------------------------------------------------------------------------- # def _encode_multipart(fields: dict[str, str], files: dict[str, Path]) -> tuple[bytes, str]: boundary = uuid.uuid4().hex out = io.BytesIO() def write(text: str) -> None: out.write(text.encode("utf-8")) for name, value in fields.items(): write(f"--{boundary}\r\n") write(f'Content-Disposition: form-data; name="{name}"\r\n\r\n') write(f"{value}\r\n") for name, path in files.items(): ctype = mimetypes.guess_type(path.name)[0] or "application/octet-stream" write(f"--{boundary}\r\n") write(f'Content-Disposition: form-data; name="{name}"; filename="{path.name}"\r\n') write(f"Content-Type: {ctype}\r\n\r\n") out.write(path.read_bytes()) write("\r\n") write(f"--{boundary}--\r\n") return out.getvalue(), f"multipart/form-data; boundary={boundary}" def _request(url: str, *, data: bytes | None = None, content_type: str | None = None, accept: str = "application/json") -> tuple[int, bytes, dict[str, str]]: req = urllib.request.Request(url, data=data, method="POST" if data else "GET") req.add_header("Accept", accept) req.add_header("User-Agent", "phage-annotation-api-client/1.0") if content_type: req.add_header("Content-Type", content_type) try: with urllib.request.urlopen(req, timeout=120) as resp: return resp.status, resp.read(), dict(resp.headers) except urllib.error.HTTPError as exc: # 4xx/5xx still carry a useful body return exc.code, exc.read(), dict(exc.headers or {}) # --------------------------------------------------------------------------- # # 1. Submit # --------------------------------------------------------------------------- # def submit(server: str, fasta: Path | None, *, gene_predictor: str = "phanotate", fast: bool = True, email: str | None = None) -> str: """Submit a job. Returns the job id.""" fields = {"gene_predictor": gene_predictor} if fast: fields["fast"] = "true" if email: fields["email"] = email files: dict[str, Path] = {} if fasta is None: fields["example"] = "1" # server-side sample genome else: files["fasta"] = fasta body, ctype = _encode_multipart(fields, files) status, payload, _ = _request(f"{server}/jobs", data=body, content_type=ctype) if status not in (200, 202): raise SystemExit(f"submission failed (HTTP {status}): {_error_of(payload)}") return json.loads(payload)["job_id"] def _error_of(payload: bytes) -> str: try: return json.loads(payload).get("error", payload[:300].decode("utf-8", "replace")) except Exception: return payload[:300].decode("utf-8", "replace") # --------------------------------------------------------------------------- # # 2. Poll # --------------------------------------------------------------------------- # def wait_for(server: str, job_id: str, *, quiet: bool = False) -> dict: """Poll until the job reaches a terminal state. Returns the final status.""" deadline = time.time() + POLL_TIMEOUT_SECONDS last = None while time.time() < deadline: status, payload, _ = _request(f"{server}/jobs/{job_id}/status.json") if status != 200: raise SystemExit(f"status check failed (HTTP {status})") info = json.loads(payload) key = (info.get("state"), info.get("stage")) if not quiet and key != last: stage = info.get("stage") or "-" print(f" {info.get('state'):<10} stage={stage:<10} {info.get('elapsed_s')}s") last = key if info.get("state") in ("completed", "failed", "expired"): return info time.sleep(POLL_SECONDS) raise SystemExit("timed out waiting for the job") # --------------------------------------------------------------------------- # # 3-4. Structured annotations, and an individual file # --------------------------------------------------------------------------- # def fetch_browser_json(server: str, job_id: str) -> dict: """Per-contig features with PHROG categories -- the same data the genome browser renders. The easiest machine-readable view of the annotations.""" status, payload, _ = _request(f"{server}/jobs/{job_id}/browser.json") if status != 200: return {"available": False} return json.loads(payload) def fetch_file(server: str, job_id: str, rel_path: str) -> bytes | None: """One result file, e.g. 'phynteny/phynteny.tsv' or 'phold/phold.gbk'.""" status, payload, _ = _request( f"{server}/jobs/{job_id}/files/{rel_path}", accept="*/*" ) return payload if status == 200 else None # --------------------------------------------------------------------------- # # 5. Full archive # --------------------------------------------------------------------------- # def download_archive(server: str, job_id: str, out_dir: Path) -> Path: status, payload, _ = _request(f"{server}/jobs/{job_id}/download", accept="*/*") if status != 200: raise SystemExit(f"download failed (HTTP {status})") out_dir.mkdir(parents=True, exist_ok=True) target = out_dir / f"{job_id}_results.zip" target.write_bytes(payload) return target # --------------------------------------------------------------------------- # # Parsing the output -- the part worth adapting # --------------------------------------------------------------------------- # def summarise_phynteny(tsv_bytes: bytes) -> None: """Parse the phynteny per-CDS table and summarise it. phynteny is the most complete view: it carries pharokka's and phold's assignments AND fills in remaining unknowns, so `phrog_category` is the annotation and `phynteny_category` is populated only where phynteny itself made a prediction. """ rows = list(csv.DictReader(io.StringIO(tsv_bytes.decode("utf-8")), delimiter="\t")) if not rows: print(" (no rows)") return categories = Counter(r.get("phrog_category", "?") for r in rows) rescued = [r for r in rows if (r.get("phynteny_category") or "NA") != "NA"] print(f" {len(rows)} CDS across {len({r.get('phage') for r in rows})} contig(s)") print(" PHROG categories:") for name, count in categories.most_common(): print(f" {count:4d} {name}") if rescued: print(f" phynteny assigned a category to {len(rescued)} previously unknown CDS:") for r in rescued[:10]: conf = r.get("phynteny_confidence", "?") print(f" {r.get('ID')} -> {r.get('phynteny_category')} (confidence {conf})") def summarise_archive(zip_path: Path) -> None: with zipfile.ZipFile(zip_path) as zf: names = zf.namelist() print(f" archive contains {len(names)} files, e.g.:") for n in sorted(names)[:8]: print(f" {n}") # --------------------------------------------------------------------------- # def main() -> None: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("fasta", nargs="?", type=Path, help="nucleotide FASTA to annotate (omit with --example)") ap.add_argument("--example", action="store_true", help="use the server's bundled sample genome instead of a file") ap.add_argument("--server", default=os.environ.get("PHAGE_SERVER", DEFAULT_SERVER)) ap.add_argument("--gene-predictor", default="phanotate", choices=["phanotate", "prodigal", "prodigal-gv"]) ap.add_argument("--no-fast", action="store_true", help="disable pharokka --fast") ap.add_argument("--email", default=None) ap.add_argument("--out", type=Path, default=Path("."), help="where to save the results archive") args = ap.parse_args() if not args.example and args.fasta is None: ap.error("give a FASTA path, or --example") if args.fasta is not None and not args.fasta.is_file(): ap.error(f"no such file: {args.fasta}") server = args.server.rstrip("/") print(f"submitting to {server} ...") job_id = submit(server, None if args.example else args.fasta, gene_predictor=args.gene_predictor, fast=not args.no_fast, email=args.email) print(f"job {job_id}\n {server}/jobs/{job_id}\n") final = wait_for(server, job_id) if final.get("state") != "completed": print(f"\njob {final.get('state')}: {final.get('error_message') or ''}") if final.get("error_hint"): print(f"hint: {final['error_hint']}") raise SystemExit(1) print("\nannotations (browser.json):") browser = fetch_browser_json(server, job_id) for contig in browser.get("contigs", [])[:5]: print(f" {contig.get('id')} {contig.get('length'):,} bp " f"{len(contig.get('features', []))} CDS") print("\nphynteny summary:") tsv = fetch_file(server, job_id, "phynteny/phynteny.tsv") if tsv: summarise_phynteny(tsv) else: print(" (phynteny.tsv not available)") print("\ndownloading archive:") archive = download_archive(server, job_id, args.out) print(f" saved {archive}") summarise_archive(archive) if __name__ == "__main__": main()