#!/usr/bin/env python3 """Server-side Open API example using only Python's standard library.""" import argparse from http.client import HTTPException import json import os from pathlib import Path import re import shutil import sys import time from urllib.error import HTTPError, URLError from urllib.parse import quote, urlsplit from urllib.request import Request, build_opener, HTTPRedirectHandler import uuid class NoRedirect(HTTPRedirectHandler): # Never forward an API key to a redirected host. def redirect_request(self, req, fp, code, msg, headers, newurl): return None class Client: def __init__(self): self.base = os.environ["ZDW_API_BASE"].rstrip("/") self.key = os.environ["ZDW_API_KEY"] target = urlsplit(self.base) if target.username or target.password or target.query or target.fragment: raise ValueError("ZDW_API_BASE must not contain credentials, a query, or a fragment") if not target.hostname or not (target.scheme == "https" or ( target.scheme == "http" and target.hostname in {"localhost", "127.0.0.1", "::1"} )): raise ValueError("Use HTTPS, or HTTP on localhost only") if any(ch in self.key for ch in "\r\n"): raise ValueError("Invalid API key") self.transport = build_opener(NoRedirect()) def request(self, method, path, body=None, idempotency_key=None): headers = {"X-Api-Key": self.key, "Accept": "application/json"} if idempotency_key: headers["Idempotency-Key"] = idempotency_key payload = None if body is not None: headers["Content-Type"] = "application/json" payload = json.dumps(body, ensure_ascii=False).encode("utf-8") request = Request(self.base + path, data=payload, headers=headers, method=method) for attempt in range(4): try: with self.transport.open(request, timeout=60) as response: return json.load(response) except HTTPError as error: if error.code in {429, 503} and attempt < 3: delay = error.headers.get("Retry-After", "3") error.close() time.sleep(min(60, max(1, int(delay) if delay.isdigit() else 3))) continue request_id = error.headers.get("X-Request-Id", "unavailable") try: code = json.load(error).get("code", "HTTP_ERROR") except (ValueError, AttributeError, HTTPException, OSError): code = "HTTP_ERROR" finally: error.close() raise RuntimeError(f"HTTP {error.code}; code={code}; requestId={request_id}") from None except (URLError, HTTPException, OSError, json.JSONDecodeError): if attempt == 3: suffix = f"; retry with Idempotency-Key={idempotency_key}" if idempotency_key else "" raise RuntimeError("Network request did not complete" + suffix) from None time.sleep(3 * (attempt + 1)) def wait(self, path): deadline, delay = time.monotonic() + 1800, 3 while time.monotonic() < deadline: task = self.request("GET", path) state = task.get("executionStatus") if state in {"succeeded", "partial", "failed", "cancelled"}: print(json.dumps({key: task.get(key) for key in ( "id", "status", "executionStatus", "coverageComplete", "sourceStatus" )}, ensure_ascii=False)) return task if state not in {"queued", "waiting_resource", "processing"}: raise RuntimeError("Unknown executionStatus; read the existing task later without creating another task") time.sleep(delay) delay = min(30, delay * 2) raise RuntimeError("Polling timeout; continue GET " + path + " later without creating another task") def submit(self, subject, keyword=None, source_query=None, idempotency_key=None): body = {} if keyword is None: body["subjectNames"] = [subject] path = "/queries" else: body.update(applicant=subject, criteria=[{"kind": "keyword", "value": keyword}]) if source_query: body["sourceRegistryQueryId"] = source_query path = "/checks" key = idempotency_key if idempotency_key is not None else str(uuid.uuid4()) if not re.fullmatch(r"[A-Za-z0-9_.:-]{1,128}", key): raise ValueError("Idempotency key must contain 1-128 letters, digits, underscores, dots, colons or hyphens") # On restart, supply --idempotency-key and the same original business arguments. # Persist this key with the exact business request before sending in production. print(f"Idempotency-Key={key}", file=sys.stderr) accepted = self.request("POST", path, body, key) tasks = accepted["queries"] if keyword is None else [accepted] for task in tasks: task_path = path + "/" + quote(task["id"], safe="") print("Accepted " + task_path, file=sys.stderr) self.wait(task_path) def download(self, document_id, destination): request = Request(self.base + "/documents/" + quote(document_id, safe="") + "/content", headers={"X-Api-Key": self.key}) output = Path(destination) created = False try: with self.transport.open(request, timeout=60) as response: # Exclusive creation prevents accidentally replacing an existing file. descriptor = os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) created = True with os.fdopen(descriptor, "wb") as target: shutil.copyfileobj(response, target) except HTTPError as error: request_id = error.headers.get("X-Request-Id", "unavailable") error.close() raise RuntimeError(f"Download failed: HTTP {error.code}; requestId={request_id}") from None except BaseException: if created: output.unlink(missing_ok=True) raise print("Download completed", file=sys.stderr) def main(): parser = argparse.ArgumentParser(description=__doc__) commands = parser.add_subparsers(dest="command", required=True) query = commands.add_parser("query") query.add_argument("subject") query.add_argument("--idempotency-key", help="Original key when recovering a request; preserve all original business arguments") check = commands.add_parser("check") check.add_argument("subject") check.add_argument("keyword") check.add_argument("--source-query") check.add_argument("--idempotency-key", help="Original key when recovering a request; preserve all original business arguments") download = commands.add_parser("download") download.add_argument("document_id") download.add_argument("destination") arguments = parser.parse_args() client = Client() if arguments.command == "download": client.download(arguments.document_id, arguments.destination) else: client.submit(arguments.subject, getattr(arguments, "keyword", None), getattr(arguments, "source_query", None), arguments.idempotency_key) if __name__ == "__main__": try: main() except (KeyError, ValueError, RuntimeError, OSError, HTTPException) as error: # Transport exceptions can contain URLs; only print controlled messages. message = str(error) if isinstance(error, (ValueError, RuntimeError)) else "Check environment variables, network, and output path" sys.exit(message)