"""HTTP access to a ShopDB instance, and the operation index it publishes. Everything the server knows about the API is fetched from the instance at runtime: the operation index, the schemas, the mount path. Nothing is generated ahead of time and nothing is pinned, so a site running an older or newer build is described accurately rather than according to whatever was true when this package was written. """ from __future__ import annotations import json import os import re import threading import time from dataclasses import dataclass from typing import Any from urllib import error, parse, request # How long a fetched index or spec stays usable before it is re-read. Long # enough that a burst of tool calls does not re-download a 700 KB document, # short enough that enabling a plugin shows up without restarting the server. CACHE_SECONDS = 300 DEFAULT_TIMEOUT = 30 # Verbs that change or destroy data. Named so the tool descriptions can say so: # a model told that DELETE is irreversible behaves better than one left to infer # it from the verb. WRITE_METHODS = {'POST', 'PUT', 'PATCH', 'DELETE'} DESTRUCTIVE_METHODS = {'DELETE'} class ShopdbError(RuntimeError): """An error worth showing the caller verbatim.""" @dataclass class Operation: method: str path: str tag: str auth: str summary: str def matches(self, needle: str) -> bool: blob = f'{self.method} {self.path} {self.tag} {self.summary}'.lower() return all(word in blob for word in needle.lower().split()) class ShopdbClient: """Talks to one ShopDB instance with one credential.""" def __init__(self, base_url: str | None = None, token: str | None = None, verify_tls: bool | None = None, timeout: int = DEFAULT_TIMEOUT): self.base_url = (base_url or os.environ.get('SHOPDB_URL') or '').rstrip('/') self.token = token or os.environ.get('SHOPDB_TOKEN') or '' if verify_tls is None: # Sites commonly run a self-signed certificate on an internal name, # so this has to be settable - but it defaults to verifying, and the # caller has to say otherwise deliberately. verify_tls = (os.environ.get('SHOPDB_VERIFY_TLS', '1').strip().lower() not in ('0', 'false', 'no')) self.verify_tls = verify_tls self.timeout = timeout self._lock = threading.Lock() self._cache: dict[str, tuple[float, Any]] = {} if not self.base_url: raise ShopdbError( 'SHOPDB_URL is not set. Point it at the instance, including any ' 'mount path, for example https://shopdb.example.net/shopdb') # ---- plumbing -------------------------------------------------------- def _opener(self): if self.verify_tls: return request.build_opener() import ssl ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE return request.build_opener(request.HTTPSHandler(context=ctx)) def request(self, method: str, path: str, params: dict | None = None, body: Any = None) -> tuple[int, Any]: """One HTTP call. Returns (status, parsed-body-or-text).""" url = self.base_url + '/' + path.lstrip('/') if params: # Drop empty values rather than sending key= , which several list # endpoints treat as a filter on the empty string. clean = {k: v for k, v in params.items() if v not in (None, '')} if clean: url += ('&' if '?' in url else '?') + parse.urlencode(clean, doseq=True) data = None headers = {'Accept': 'application/json'} if self.token: headers['Authorization'] = f'Bearer {self.token}' if body is not None: data = json.dumps(body).encode('utf-8') headers['Content-Type'] = 'application/json' req = request.Request(url, data=data, headers=headers, method=method.upper()) try: with self._opener().open(req, timeout=self.timeout) as response: return response.status, _decode(response.read(), response.headers.get('Content-Type')) except error.HTTPError as exc: # An error body from this API is the useful part (it carries the # error code and message), so return it rather than raising. return exc.code, _decode(exc.read(), exc.headers.get('Content-Type')) except error.URLError as exc: raise ShopdbError(f'could not reach {self.base_url}: {exc.reason}') from exc def _cached(self, key: str, fetch): with self._lock: hit = self._cache.get(key) if hit and (time.time() - hit[0]) < CACHE_SECONDS: return hit[1] value = fetch() with self._lock: self._cache[key] = (time.time(), value) return value # ---- what the instance publishes about itself ------------------------ def operations(self) -> list[Operation]: """The published operation index, parsed. Read from /api/docs/operations.txt rather than the full spec: the index is around 70 KB against 700 KB, and listing what exists does not need the schemas. """ def fetch(): status, text = self.request('GET', '/api/docs/operations.txt') if status != 200: raise ShopdbError( f'operation index unavailable (HTTP {status}). This instance ' 'may predate /api/docs/operations.txt.') return _parse_index(text if isinstance(text, str) else str(text)) return self._cached('operations', fetch) def spec(self, tag: str | None = None) -> dict: """The OpenAPI spec, optionally narrowed to one tag group.""" def fetch(): path = '/api/docs/openapi.json' status, payload = self.request('GET', path, params={'tag': tag} if tag else None) if status != 200: raise ShopdbError(f'spec unavailable (HTTP {status}): {payload}') return payload return self._cached(f'spec:{tag or "*"}', fetch) def find_operation(self, method: str, path: str) -> dict | None: """The spec entry for one operation, fetched by its tag to stay small.""" method = method.lower() for op in self.operations(): if op.method.lower() == method and op.path == path: spec = self.spec(op.tag) return (spec.get('paths', {}).get(path, {}) or {}).get(method) return None _INDEX_LINE = re.compile( r'^(?P[A-Z]+)\s+(?P\S+)\s+\[(?P[^\]]*)\]\s+' r'\((?P[^)]*)\)\s*(?P.*)$') def _parse_index(text: str) -> list[Operation]: out = [] for line in text.splitlines(): line = line.strip() if not line or line.startswith('#'): continue m = _INDEX_LINE.match(line) if m: out.append(Operation(m.group('method'), m.group('path'), m.group('tag'), m.group('auth'), m.group('summary').strip())) return out def _decode(raw: bytes, content_type: str | None): text = raw.decode('utf-8', errors='replace') if content_type and 'json' in content_type.lower(): try: return json.loads(text) except json.JSONDecodeError: return text return text