From 36640d5d06234d0620c5d5bb004b1c8fac787a72 Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 09:09:52 -0400 Subject: [PATCH 01/10] OpenConceptLab/ocl_online#507 | Fix lint and add CI (ruff + pytest) Co-Authored-By: Claude Haiku 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- .github/workflows/ci.yml | 21 +++++++++++++++++++++ src/ocl_cli/api_client.py | 4 ++-- src/ocl_cli/commands/cascade.py | 2 +- src/ocl_cli/commands/expansion.py | 2 +- src/ocl_cli/commands/mapping.py | 4 +++- src/ocl_cli/commands/org.py | 7 +++++-- src/ocl_cli/commands/ref.py | 2 +- src/ocl_cli/commands/repo.py | 7 +++++-- src/ocl_cli/commands/server.py | 2 +- src/ocl_cli/commands/task.py | 3 +-- src/ocl_cli/commands/user.py | 7 +++++-- src/ocl_cli/config.py | 1 - src/ocl_cli/main.py | 12 ++++++------ src/ocl_cli/output.py | 2 -- 14 files changed, 52 insertions(+), 24 deletions(-) create mode 100644 .github/workflows/ci.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..d582821 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,21 @@ +name: CI + +on: + pull_request: + push: + branches: [main] + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.10", "3.13"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - run: pip install -e ".[dev]" + - run: ruff check src tests + - run: pytest -q diff --git a/src/ocl_cli/api_client.py b/src/ocl_cli/api_client.py index 6c805d7..8521121 100644 --- a/src/ocl_cli/api_client.py +++ b/src/ocl_cli/api_client.py @@ -9,8 +9,6 @@ from typing import Any, Optional import httpx - -from ocl_cli import __version__ from tenacity import ( retry, retry_if_exception_type, @@ -18,6 +16,8 @@ wait_exponential, ) +from ocl_cli import __version__ + class APIError(Exception): """Structured API error with status code and detail.""" diff --git a/src/ocl_cli/commands/cascade.py b/src/ocl_cli/commands/cascade.py index 7c55af2..25c3d54 100644 --- a/src/ocl_cli/commands/cascade.py +++ b/src/ocl_cli/commands/cascade.py @@ -4,7 +4,7 @@ from ocl_cli.api_client import APIError from ocl_cli.main import handle_api_error -from ocl_cli.output import output_result, format_cascade_results +from ocl_cli.output import format_cascade_results, output_result @click.command("cascade") diff --git a/src/ocl_cli/commands/expansion.py b/src/ocl_cli/commands/expansion.py index 5ae558a..5daf7e3 100644 --- a/src/ocl_cli/commands/expansion.py +++ b/src/ocl_cli/commands/expansion.py @@ -4,7 +4,7 @@ from ocl_cli.api_client import APIError from ocl_cli.main import handle_api_error -from ocl_cli.output import output_result, format_expansion_list, format_expansion_detail +from ocl_cli.output import format_expansion_detail, format_expansion_list, output_result @click.group() diff --git a/src/ocl_cli/commands/mapping.py b/src/ocl_cli/commands/mapping.py index 7060fce..ccd6960 100644 --- a/src/ocl_cli/commands/mapping.py +++ b/src/ocl_cli/commands/mapping.py @@ -7,8 +7,10 @@ from ocl_cli.api_client import APIError from ocl_cli.main import handle_api_error from ocl_cli.output import ( - output_result, format_mapping_list, format_mapping_detail, + format_mapping_detail, + format_mapping_list, format_version_list, + output_result, ) diff --git a/src/ocl_cli/commands/org.py b/src/ocl_cli/commands/org.py index 5ca336d..dbf5147 100644 --- a/src/ocl_cli/commands/org.py +++ b/src/ocl_cli/commands/org.py @@ -8,8 +8,11 @@ from ocl_cli.api_client import APIError from ocl_cli.main import handle_api_error from ocl_cli.output import ( - output_result, format_org_list, format_org_detail, - format_member_list, format_repo_list, + format_member_list, + format_org_detail, + format_org_list, + format_repo_list, + output_result, ) diff --git a/src/ocl_cli/commands/ref.py b/src/ocl_cli/commands/ref.py index 2425187..7f68da6 100644 --- a/src/ocl_cli/commands/ref.py +++ b/src/ocl_cli/commands/ref.py @@ -4,7 +4,7 @@ from ocl_cli.api_client import APIError from ocl_cli.main import handle_api_error -from ocl_cli.output import output_result, format_reference_list +from ocl_cli.output import format_reference_list, output_result @click.group() diff --git a/src/ocl_cli/commands/repo.py b/src/ocl_cli/commands/repo.py index 3400ca4..4a66e7d 100644 --- a/src/ocl_cli/commands/repo.py +++ b/src/ocl_cli/commands/repo.py @@ -8,8 +8,11 @@ from ocl_cli.api_client import APIError from ocl_cli.main import handle_api_error from ocl_cli.output import ( - output_result, format_repo_list, format_repo_detail, - format_version_list, format_extras, + format_extras, + format_repo_detail, + format_repo_list, + format_version_list, + output_result, ) diff --git a/src/ocl_cli/commands/server.py b/src/ocl_cli/commands/server.py index 9013a06..6b92a3a 100644 --- a/src/ocl_cli/commands/server.py +++ b/src/ocl_cli/commands/server.py @@ -2,7 +2,7 @@ import click -from ocl_cli.output import output_result, format_table +from ocl_cli.output import format_table, output_result @click.group() diff --git a/src/ocl_cli/commands/task.py b/src/ocl_cli/commands/task.py index 465f895..d8a4ece 100644 --- a/src/ocl_cli/commands/task.py +++ b/src/ocl_cli/commands/task.py @@ -1,12 +1,11 @@ """Task commands: list and get async tasks.""" -import sys import click from ocl_cli.api_client import APIError from ocl_cli.main import handle_api_error -from ocl_cli.output import output_result, format_task_list, format_task_detail +from ocl_cli.output import format_task_detail, format_task_list, output_result @click.group() diff --git a/src/ocl_cli/commands/user.py b/src/ocl_cli/commands/user.py index 7dda207..6626a0c 100644 --- a/src/ocl_cli/commands/user.py +++ b/src/ocl_cli/commands/user.py @@ -5,8 +5,11 @@ from ocl_cli.api_client import APIError from ocl_cli.main import handle_api_error from ocl_cli.output import ( - output_result, format_user_list, format_user_detail, - format_org_list, format_repo_list, + format_org_list, + format_repo_list, + format_user_detail, + format_user_list, + output_result, ) diff --git a/src/ocl_cli/config.py b/src/ocl_cli/config.py index b1e9399..714f9b3 100644 --- a/src/ocl_cli/config.py +++ b/src/ocl_cli/config.py @@ -10,7 +10,6 @@ from pathlib import Path from typing import Optional - CONFIG_DIR = Path.home() / ".ocl" CONFIG_FILE = CONFIG_DIR / "config.json" diff --git a/src/ocl_cli/main.py b/src/ocl_cli/main.py index f0766dd..697c679 100644 --- a/src/ocl_cli/main.py +++ b/src/ocl_cli/main.py @@ -64,14 +64,14 @@ def handle_api_error(e: APIError) -> None: cli.add_command(server) # Phase 2: Read commands -from ocl_cli.commands.org import org # noqa: E402 -from ocl_cli.commands.user import user # noqa: E402 -from ocl_cli.commands.repo import repo # noqa: E402 +from ocl_cli.commands.cascade import cascade_cmd # noqa: E402 from ocl_cli.commands.concept import concept # noqa: E402 +from ocl_cli.commands.expansion import expansion # noqa: E402 from ocl_cli.commands.mapping import mapping # noqa: E402 -from ocl_cli.commands.cascade import cascade_cmd # noqa: E402 +from ocl_cli.commands.org import org # noqa: E402 from ocl_cli.commands.ref import ref # noqa: E402 -from ocl_cli.commands.expansion import expansion # noqa: E402 +from ocl_cli.commands.repo import repo # noqa: E402 +from ocl_cli.commands.user import user # noqa: E402 cli.add_command(org) cli.add_command(user) @@ -88,8 +88,8 @@ def handle_api_error(e: APIError) -> None: cli.add_command(task) # Operational tools -from ocl_cli.commands.tool import tool # noqa: E402 from ocl_cli.commands.import_cmd import import_cmd # noqa: E402 +from ocl_cli.commands.tool import tool # noqa: E402 cli.add_command(tool) cli.add_command(import_cmd) diff --git a/src/ocl_cli/output.py b/src/ocl_cli/output.py index ed425b7..c149530 100644 --- a/src/ocl_cli/output.py +++ b/src/ocl_cli/output.py @@ -4,7 +4,6 @@ """ import json -import sys from typing import Any, Callable, Optional import click @@ -878,7 +877,6 @@ def _format_runtime(started: Any, finished: Any) -> str: return "" try: from datetime import datetime - fmt = "%Y-%m-%dT%H:%M:%S" s = str(started)[:19] f = str(finished)[:19] delta = datetime.fromisoformat(s) - datetime.fromisoformat(f) From a4b96df274fb803c3c250d72d3c9efc6665275ea Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 09:11:21 -0400 Subject: [PATCH 02/10] OpenConceptLab/ocl_online#507 | Move the API client into the openconceptlab core package Co-Authored-By: Claude Haiku 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- pyproject.toml | 2 +- src/ocl_cli/__init__.py | 6 +- src/ocl_cli/api_client.py | 1665 +------------------------------- src/openconceptlab/__init__.py | 12 + src/openconceptlab/client.py | 1663 +++++++++++++++++++++++++++++++ tests/test_core_package.py | 37 + 6 files changed, 1721 insertions(+), 1664 deletions(-) create mode 100644 src/openconceptlab/__init__.py create mode 100644 src/openconceptlab/client.py create mode 100644 tests/test_core_package.py diff --git a/pyproject.toml b/pyproject.toml index c2f76d6..d05f893 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "openconceptlab" -version = "0.1.1" +version = "0.2.0" description = "Command-line interface for the Open Concept Lab API" readme = "README.md" license = "MPL-2.0" diff --git a/src/ocl_cli/__init__.py b/src/ocl_cli/__init__.py index c0afc63..ae61658 100644 --- a/src/ocl_cli/__init__.py +++ b/src/ocl_cli/__init__.py @@ -1,3 +1,7 @@ """OCL CLI - Command-line interface for the Open Concept Lab API.""" -__version__ = "0.1.0" +from openconceptlab import __version__ + +CLIENT_NAME = f"ocl-cli/{__version__}" + +__all__ = ["CLIENT_NAME", "__version__"] diff --git a/src/ocl_cli/api_client.py b/src/ocl_cli/api_client.py index 8521121..cb01028 100644 --- a/src/ocl_cli/api_client.py +++ b/src/ocl_cli/api_client.py @@ -1,1664 +1,5 @@ -"""Synchronous OCL API client. +"""Compatibility shim: the client lives in the open core, `openconceptlab.client`.""" -Adapted from ocl-mcp's async client. Uses httpx.Client (sync) with retry logic. -Business logic lives here so it can be reused by future MCP server wrappers. -""" +from openconceptlab.client import APIError, OCLAPIClient -import platform -import sys -from typing import Any, Optional - -import httpx -from tenacity import ( - retry, - retry_if_exception_type, - stop_after_attempt, - wait_exponential, -) - -from ocl_cli import __version__ - - -class APIError(Exception): - """Structured API error with status code and detail.""" - - def __init__(self, message: str, status_code: int = 0, detail: str = ""): - super().__init__(message) - self.status_code = status_code - self.detail = detail - - def to_dict(self) -> dict: - d = {"error": str(self), "status_code": self.status_code} - if self.detail: - d["detail"] = self.detail - return d - - -def _repo_type_stem(repo_type: str) -> str: - """Convert 'source'/'collection' to plural URL segment.""" - if repo_type == "source": - return "sources" - elif repo_type == "collection": - return "collections" - else: - raise ValueError(f"repo_type must be 'source' or 'collection', got '{repo_type}'") - - -def _validate_owner_type(owner_type: str) -> None: - if owner_type not in ("users", "orgs"): - raise ValueError(f"owner_type must be 'users' or 'orgs', got '{owner_type}'") - - -def _build_repo_endpoint( - owner_type: str, - owner: str, - repo_type: str, - repo: str, - repo_version: Optional[str] = None, - suffix: str = "", -) -> str: - """Build a repository-scoped endpoint URL.""" - _validate_owner_type(owner_type) - stem = _repo_type_stem(repo_type) - base = f"/{owner_type}/{owner}/{stem}/{repo}" - if repo_version: - base = f"{base}/{repo_version}" - return f"{base}/{suffix}" if suffix else f"{base}/" - - -class OCLAPIClient: - """Synchronous client for the OCL API.""" - - DEFAULT_BASE_URL = "https://api.openconceptlab.org" - - def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None): - self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/") - self.token = token - self.debug = False - self.show_request = False - self._show_request_server_printed = False - - headers = { - "Accept": "application/json", - "X-OCL-CLIENT": f"ocl-cli/{__version__}", - "User-Agent": f"ocl-cli/{__version__} (httpx/{httpx.__version__}; python/{platform.python_version()})", - } - if self.token: - headers["Authorization"] = f"Token {self.token}" - - self.client = httpx.Client( - base_url=self.base_url, - headers=headers, - timeout=30.0, - limits=httpx.Limits(max_keepalive_connections=10, max_connections=20), - ) - - def close(self): - self.client.close() - - def __enter__(self): - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - self.close() - - def _log_request(self, method: str, endpoint: str, params: dict | None = None, - body: Any = None): - if self.debug: - url = f"{self.base_url}{endpoint}" - if params: - from urllib.parse import urlencode - url = f"{url}?{urlencode(params, doseq=True)}" - print(f" {method} {url}", file=sys.stderr) - if body is not None: - import json as _json - print(f" Body: {_json.dumps(body, default=str)}", file=sys.stderr) - if self.show_request: - if not self._show_request_server_printed: - print(f"Server: {self.base_url}", file=sys.stderr) - self._show_request_server_printed = True - url = endpoint - if params: - from urllib.parse import urlencode - url = f"{url}?{urlencode(params, doseq=True)}" - print(f" → {method} {url}", file=sys.stderr) - - def _handle_error(self, response: httpx.Response) -> None: - """Convert HTTP errors to APIError.""" - if response.status_code == 429: - retry_after = response.headers.get("Retry-After", "60") - raise APIError( - f"Rate limit exceeded. Retry after {retry_after}s", - status_code=429, - ) - if response.status_code >= 400: - try: - detail = response.json() - except Exception: - detail = response.text - raise APIError( - f"HTTP {response.status_code}", - status_code=response.status_code, - detail=str(detail), - ) - - @retry( - stop=stop_after_attempt(3), - wait=wait_exponential(multiplier=1, min=4, max=60), - retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), - ) - def get(self, endpoint: str, params: Optional[dict] = None, timeout: Optional[float] = None) -> Any: - self._log_request("GET", endpoint, params) - kwargs: dict[str, Any] = {"params": params} - if timeout is not None: - kwargs["timeout"] = timeout - response = self.client.get(endpoint, **kwargs) - self._handle_error(response) - return response.json() - - @retry( - stop=stop_after_attempt(3), - wait=wait_exponential(multiplier=1, min=4, max=60), - retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), - ) - def post(self, endpoint: str, json: Optional[dict] = None, params: Optional[dict] = None) -> Any: - self._log_request("POST", endpoint, params, body=json) - response = self.client.post(endpoint, json=json, params=params) - self._handle_error(response) - return response.json() - - @retry( - stop=stop_after_attempt(3), - wait=wait_exponential(multiplier=1, min=4, max=60), - retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), - ) - def put(self, endpoint: str, json: Optional[dict] = None, params: Optional[dict] = None) -> Any: - self._log_request("PUT", endpoint, params, body=json) - response = self.client.put(endpoint, json=json, params=params) - self._handle_error(response) - return response.json() - - @retry( - stop=stop_after_attempt(3), - wait=wait_exponential(multiplier=1, min=4, max=60), - retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), - ) - def patch(self, endpoint: str, json: Optional[dict] = None, params: Optional[dict] = None) -> Any: - self._log_request("PATCH", endpoint, params, body=json) - response = self.client.patch(endpoint, json=json, params=params) - self._handle_error(response) - return response.json() - - @retry( - stop=stop_after_attempt(3), - wait=wait_exponential(multiplier=1, min=4, max=60), - retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), - ) - def delete(self, endpoint: str, params: Optional[dict] = None) -> Any: - self._log_request("DELETE", endpoint, params) - response = self.client.delete(endpoint, params=params) - self._handle_error(response) - if response.status_code == 204: - return {} - return response.json() - - def request( - self, - method: str, - endpoint: str, - params: Optional[dict] = None, - json: Optional[dict] = None, - follow_redirects: bool = False, - timeout: Optional[float] = None, - ) -> httpx.Response: - """Low-level request returning the raw httpx.Response. - - Unlike get/post/delete, this does NOT call _handle_error(). - Callers are responsible for interpreting status codes. - Used for APIs with non-standard status code semantics (e.g. exports). - """ - self._log_request(method, endpoint, params, body=json) - kwargs: dict[str, Any] = {"params": params, "follow_redirects": follow_redirects} - if json is not None: - kwargs["json"] = json - if timeout is not None: - kwargs["timeout"] = timeout - return self.client.request(method, endpoint, **kwargs) - - # ── Helpers ────────────────────────────────────────────────────── - - def _normalize(self, response: Any, headers: Optional[dict] = None) -> dict: - """Normalize list responses to {count, results} format. - - When ``headers`` are provided (from _get_list), ``num_found`` is used - as the authoritative total count for pagination. - """ - num_found = int(headers.get("num_found", 0)) if headers else 0 - if isinstance(response, list): - return {"count": num_found or len(response), "results": response} - if isinstance(response, dict) and num_found: - response["count"] = num_found - return response - - @retry( - stop=stop_after_attempt(3), - wait=wait_exponential(multiplier=1, min=4, max=60), - retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), - ) - def _get_list(self, endpoint: str, params: Optional[dict] = None) -> dict: - """GET a list endpoint, preserving pagination headers. - - The OCL API returns lists with pagination metadata in response - headers (num_found, num_returned, page_number). The standard - get() discards headers, so this method captures them and feeds - them to _normalize() for correct total counts. - """ - self._log_request("GET", endpoint, params) - response = self.client.get(endpoint, params=params) - self._handle_error(response) - return self._normalize(response.json(), headers=dict(response.headers)) - - def _require_auth(self) -> None: - if not self.token: - raise APIError("Authentication required. Run 'ocl login' first.", status_code=401) - - # ── High-level API methods ─────────────────────────────────────── - - def get_user(self) -> dict: - """Get current authenticated user info.""" - self._require_auth() - return self.get("/user/") - - # ── Organization operations ──────────────────────────────────── - - def list_orgs( - self, - query: Optional[str] = None, - verbose: bool = False, - limit: int = 25, - page: int = 1, - ) -> dict: - """List/search organizations.""" - params: dict[str, Any] = {"limit": limit, "page": page} - if query: - params["q"] = query - if verbose: - params["verbose"] = True - return self._get_list("/orgs/", params=params) - - def get_org(self, org: str) -> dict: - """Get a specific organization.""" - return self.get(f"/orgs/{org}/") - - def list_org_repos( - self, - org: str, - repo_type: str = "all", - limit: int = 20, - page: int = 1, - ) -> dict: - """List repositories owned by an organization.""" - params: dict[str, Any] = {"limit": limit, "page": page} - if repo_type != "all": - params["repoType"] = "Source" if repo_type == "source" else "Collection" - return self._get_list(f"/orgs/{org}/repos/", params=params) - - # ── User operations ────────────────────────────────────────── - - def list_users( - self, - query: Optional[str] = None, - verbose: bool = False, - limit: int = 25, - page: int = 1, - ) -> dict: - """List/search users.""" - params: dict[str, Any] = {"limit": limit, "page": page} - if query: - params["q"] = query - if verbose: - params["verbose"] = True - return self._get_list("/users/", params=params) - - def get_user_detail(self, username: str) -> dict: - """Get a specific user by username.""" - return self.get(f"/users/{username}/") - - def list_user_repos( - self, - username: str, - repo_type: str = "all", - limit: int = 20, - page: int = 1, - ) -> dict: - """List repositories owned by a user.""" - params: dict[str, Any] = {"limit": limit, "page": page} - if repo_type != "all": - params["repoType"] = "Source" if repo_type == "source" else "Collection" - return self._get_list(f"/users/{username}/repos/", params=params) - - def list_user_orgs(self, username: str, limit: int = 100) -> dict: - """List organizations a user belongs to.""" - return self._get_list(f"/users/{username}/orgs/", params={"limit": limit, "verbose": "true"}) - - def get_org_members(self, org: str, limit: int = 100) -> dict: - """List members of an organization.""" - return self._get_list(f"/orgs/{org}/members/", params={"limit": limit}) - - def search_repos( - self, - query: Optional[str] = None, - owner: Optional[str] = None, - owner_type: str = "all", - repo_type: str = "all", - custom_validation_schema: Optional[str] = None, - updated_since: Optional[str] = None, - all_versions: bool = False, - verbose: bool = False, - limit: int = 20, - page: int = 1, - ) -> dict: - """Search for repositories (sources and collections).""" - params: dict[str, Any] = {"limit": limit, "page": page} - if verbose: - params["verbose"] = True - if query: - params["q"] = query - if repo_type != "all": - params["repoType"] = "Source" if repo_type == "source" else "Collection" - if custom_validation_schema: - params["customValidationSchema"] = custom_validation_schema - if updated_since: - params["updatedSince"] = updated_since - if all_versions: - params["allVersions"] = "true" - - # Use scoped endpoint when owner is specified - if owner and owner_type and owner_type != "all": - ot = owner_type # "orgs" or "users" - endpoint = f"/{ot}/{owner}/repos/" - elif owner: - # Default to orgs scope - endpoint = f"/orgs/{owner}/repos/" - else: - # Global search — add owner/ownerType as query params - endpoint = "/repos/" - if owner_type != "all": - params["ownerType"] = "Organization" if owner_type == "orgs" else "User" - - return self._get_list(endpoint, params=params) - - def get_repo( - self, - owner: str, - repo: str, - owner_type: str = "orgs", - repo_type: str = "source", - repo_version: Optional[str] = None, - include_summary: bool = False, - ) -> dict: - """Get a specific repository, optionally at a version.""" - endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo, repo_version) - params = {} - if include_summary: - params["includeSummary"] = "true" - return self.get(endpoint, params=params or None) - - def get_repo_versions( - self, - owner: str, - repo: str, - owner_type: str = "orgs", - repo_type: str = "source", - released: Optional[bool] = None, - processing: Optional[bool] = None, - updated_since: Optional[str] = None, - limit: int = 20, - page: int = 1, - ) -> dict: - """List versions for a repository.""" - endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo, suffix="versions/") - params: dict[str, Any] = {"limit": limit, "page": page} - if released is not None: - params["released"] = str(released).lower() - if processing is not None: - params["processing"] = str(processing).lower() - if updated_since: - params["updatedSince"] = updated_since - return self._get_list(endpoint, params=params) - - def search_concepts( - self, - query: Optional[str] = None, - owner: Optional[str] = None, - owner_type: Optional[str] = None, - repo_type: Optional[str] = None, - repo: Optional[str] = None, - repo_version: Optional[str] = None, - concept_class: Optional[str] = None, - datatype: Optional[str] = None, - locale: Optional[str] = None, - include_retired: bool = False, - include_mappings: bool = False, - include_inverse_mappings: bool = False, - updated_since: Optional[str] = None, - sort: Optional[str] = None, - verbose: bool = False, - limit: int = 20, - page: int = 1, - ) -> dict: - """Search for concepts globally or within a repository.""" - params: dict[str, Any] = {"limit": limit, "page": page} - if verbose: - params["verbose"] = True - if query: - params["q"] = query - if concept_class: - params["conceptClass"] = concept_class - if datatype: - params["datatype"] = datatype - if locale: - params["locale"] = locale - if include_retired: - params["includeRetired"] = "true" - if include_mappings: - params["includeMappings"] = "true" - if include_inverse_mappings: - params["includeInverseMappings"] = "true" - if updated_since: - params["updatedSince"] = updated_since - if sort: - params["sortAsc"] = sort if not sort.startswith("-") else None - if sort.startswith("-"): - params["sortDesc"] = sort[1:] - # Determine endpoint scope — use narrowest available - if all([owner_type, owner, repo_type, repo]): - endpoint = _build_repo_endpoint( - owner_type, owner, repo_type, repo, repo_version, suffix="concepts/" - ) - elif owner and repo: - ot = owner_type or "orgs" - rt = repo_type or "source" - endpoint = _build_repo_endpoint(ot, owner, rt, repo, repo_version, suffix="concepts/") - else: - # Global search — add owner/ownerType as query params - endpoint = "/concepts/" - if owner: - params["owner"] = owner - if owner_type and owner_type not in ("all",): - params["ownerType"] = "Organization" if owner_type == "orgs" else "User" - - return self._get_list(endpoint, params=params) - - def get_concept( - self, - owner: str, - source: str, - concept_id: str, - owner_type: str = "orgs", - repo_version: Optional[str] = None, - concept_version: Optional[str] = None, - include_mappings: bool = False, - include_inverse_mappings: bool = False, - verbose: bool = False, - ) -> dict: - """Get a single concept.""" - _validate_owner_type(owner_type) - stem = _repo_type_stem("source") - base = f"/{owner_type}/{owner}/{stem}/{source}" - if repo_version: - base = f"{base}/{repo_version}" - endpoint = f"{base}/concepts/{concept_id}/" - if concept_version: - endpoint = f"{base}/concepts/{concept_id}/{concept_version}/" - - params: dict[str, Any] = {} - if include_mappings: - params["includeMappings"] = "true" - if include_inverse_mappings: - params["includeInverseMappings"] = "true" - if verbose: - params["verbose"] = "true" - - return self.get(endpoint, params=params or None) - - def get_concept_versions( - self, - owner: str, - source: str, - concept_id: str, - owner_type: str = "orgs", - limit: int = 20, - page: int = 1, - ) -> dict: - """List version history for a concept.""" - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/versions/" - return self._get_list(endpoint, params={"limit": limit, "page": page}) - - def get_concept_names( - self, - owner: str, - source: str, - concept_id: str, - owner_type: str = "orgs", - ) -> dict: - """List names for a concept.""" - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/names/" - return self._get_list(endpoint) - - def get_concept_descriptions( - self, - owner: str, - source: str, - concept_id: str, - owner_type: str = "orgs", - ) -> dict: - """List descriptions for a concept.""" - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/descriptions/" - return self._get_list(endpoint) - - def get_concept_extras( - self, - owner: str, - source: str, - concept_id: str, - owner_type: str = "orgs", - ) -> dict: - """Get extras (custom attributes) for a concept.""" - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/extras/" - return self.get(endpoint) - - def search_mappings( - self, - query: Optional[str] = None, - owner: Optional[str] = None, - owner_type: Optional[str] = None, - repo_type: Optional[str] = None, - repo: Optional[str] = None, - repo_version: Optional[str] = None, - map_type: Optional[str] = None, - from_source: Optional[str] = None, - from_concept: Optional[str] = None, - from_concept_owner: Optional[str] = None, - to_source: Optional[str] = None, - to_concept: Optional[str] = None, - to_concept_owner: Optional[str] = None, - include_retired: bool = False, - updated_since: Optional[str] = None, - sort: Optional[str] = None, - verbose: bool = False, - limit: int = 20, - page: int = 1, - ) -> dict: - """Search for mappings globally or within a repository.""" - params: dict[str, Any] = {"limit": limit, "page": page} - if verbose: - params["verbose"] = True - if query: - params["q"] = query - if map_type: - params["mapType"] = map_type - if from_source: - params["fromConceptSource"] = from_source - if from_concept: - params["fromConcept"] = from_concept - if from_concept_owner: - params["fromConceptOwner"] = from_concept_owner - if to_source: - params["toConceptSource"] = to_source - if to_concept: - params["toConcept"] = to_concept - if to_concept_owner: - params["toConceptOwner"] = to_concept_owner - if include_retired: - params["includeRetired"] = "true" - if updated_since: - params["updatedSince"] = updated_since - if sort: - params["sortAsc"] = sort if not sort.startswith("-") else None - if sort.startswith("-"): - params["sortDesc"] = sort[1:] - # Determine endpoint scope — use narrowest available - if all([owner_type, owner, repo_type, repo]): - endpoint = _build_repo_endpoint( - owner_type, owner, repo_type, repo, repo_version, suffix="mappings/" - ) - elif owner and repo: - ot = owner_type or "orgs" - rt = repo_type or "source" - endpoint = _build_repo_endpoint(ot, owner, rt, repo, repo_version, suffix="mappings/") - else: - # Global search — add owner/ownerType as query params - endpoint = "/mappings/" - if owner: - params["owner"] = owner - if owner_type and owner_type not in ("all",): - params["ownerType"] = "Organization" if owner_type == "orgs" else "User" - - return self._get_list(endpoint, params=params) - - def get_mapping( - self, - owner: str, - source: str, - mapping_id: str, - owner_type: str = "orgs", - repo_version: Optional[str] = None, - ) -> dict: - """Get a single mapping.""" - _validate_owner_type(owner_type) - base = f"/{owner_type}/{owner}/sources/{source}" - if repo_version: - base = f"{base}/{repo_version}" - return self.get(f"{base}/mappings/{mapping_id}/") - - def get_mapping_versions( - self, - owner: str, - source: str, - mapping_id: str, - owner_type: str = "orgs", - limit: int = 20, - page: int = 1, - ) -> dict: - """List version history for a mapping.""" - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/mappings/{mapping_id}/versions/" - return self._get_list(endpoint, params={"limit": limit, "page": page}) - - def cascade( - self, - owner: str, - repo: str, - concept_id: str, - owner_type: str = "orgs", - repo_type: str = "source", - repo_version: Optional[str] = None, - map_types: Optional[list[str]] = None, - exclude_map_types: Optional[list[str]] = None, - return_map_types: Optional[list[str]] = None, - method: str = "sourcetoconcepts", - cascade_hierarchy: bool = True, - cascade_mappings: bool = True, - cascade_levels: str = "*", - reverse: bool = False, - view: str = "hierarchy", - verbose: bool = False, - omit_if_exists_in: Optional[str] = None, - equivalency_map_type: Optional[str] = None, - ) -> dict: - """Execute $cascade operation on a concept.""" - endpoint = _build_repo_endpoint( - owner_type, owner, repo_type, repo, repo_version, - suffix=f"concepts/{concept_id}/$cascade/" - ) - - params: dict[str, Any] = {} - if verbose: - params["verbose"] = True - if map_types: - params["mapTypes"] = ",".join(map_types) - if exclude_map_types: - params["excludeMapTypes"] = ",".join(exclude_map_types) - if return_map_types: - params["returnMapTypes"] = ",".join(return_map_types) - if method != "sourcetoconcepts": - params["method"] = method - if not cascade_hierarchy: - params["cascadeHierarchy"] = "false" - if not cascade_mappings: - params["cascadeMappings"] = "false" - if cascade_levels != "*": - params["cascadeLevels"] = cascade_levels - if reverse: - params["reverse"] = "true" - if view != "hierarchy": - params["view"] = view - else: - params["view"] = "hierarchy" - if omit_if_exists_in: - params["omitIfExistsIn"] = omit_if_exists_in - if equivalency_map_type: - params["equivalencyMapType"] = equivalency_map_type - - result = self.get(endpoint, params=params or None, timeout=120.0) - if isinstance(result, list): - return {"resourceType": "Bundle", "type": "searchset", "entry": result} - return result - - def match_concepts( - self, - terms: list[str], - target_repo_url: str, - concept_class: Optional[str] = None, - datatype: Optional[str] = None, - limit: int = 5, - include_retired: bool = False, - include_mappings: bool = False, - semantic: bool = True, - verbose: bool = False, - ) -> dict: - """Match concepts using the $match endpoint.""" - rows = [{"name": term} for term in terms] - body: dict[str, Any] = { - "rows": rows, - "target_repo_url": target_repo_url, - } - filters: dict[str, Any] = {} - if concept_class: - filters["concept_class"] = concept_class - if datatype: - filters["datatype"] = datatype - if filters: - body["filter"] = filters - params: dict[str, Any] = { - "includeSearchMeta": True, - "semantic": semantic, - "bestMatch": True, - } - if limit: - params["limit"] = limit - if include_retired: - params["includeRetired"] = True - if include_mappings: - params["includeMappings"] = True - if verbose: - params["verbose"] = True - - result = self.post("/concepts/$match/", params=params, json=body) - return self._normalize(result) - - def list_expansions( - self, - owner: str, - collection: str, - collection_version: str, - owner_type: str = "orgs", - ) -> dict: - """List expansions for a collection version.""" - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/collections/{collection}/{collection_version}/expansions/" - return self._get_list(endpoint, params={"verbose": True}) - - def get_expansion( - self, - owner: str, - collection: str, - owner_type: str = "orgs", - collection_version: Optional[str] = None, - expansion_id: Optional[str] = None, - ) -> dict: - """Get a specific or default expansion for a collection.""" - _validate_owner_type(owner_type) - - if expansion_id: - if collection_version: - endpoint = f"/{owner_type}/{owner}/collections/{collection}/{collection_version}/expansions/{expansion_id}/" - else: - endpoint = f"/{owner_type}/{owner}/collections/{collection}/expansions/{expansion_id}/" - return self.get(endpoint, params={"verbose": True}) - - # Look up default expansion from collection version - if collection_version: - col_endpoint = f"/{owner_type}/{owner}/collections/{collection}/{collection_version}/" - else: - col_endpoint = f"/{owner_type}/{owner}/collections/{collection}/" - - col_data = self.get(col_endpoint) - expansion_url = col_data.get("expansion_url") - if expansion_url: - return self.get(expansion_url, params={"verbose": True}) - return { - "message": "No default expansion configured for this collection version", - "expansion_url": None, - } - - def list_collection_refs( - self, - owner: str, - collection: str, - owner_type: str = "orgs", - collection_version: Optional[str] = None, - limit: int = 20, - page: int = 1, - ) -> dict: - """List references in a collection.""" - _validate_owner_type(owner_type) - base = f"/{owner_type}/{owner}/collections/{collection}" - if collection_version: - base = f"{base}/{collection_version}" - endpoint = f"{base}/references/" - return self._get_list(endpoint, params={"limit": limit, "page": page}) - - def resolve_reference(self, *expressions: str, namespace: Optional[str] = None) -> dict: - """Resolve references using $resolveReference.""" - params: dict[str, Any] = {} - if namespace: - params["namespace"] = namespace - body = {"references": list(expressions)} - return self.post("/$resolveReference/", json=body, params=params or None) - - # ── Write operations ───────────────────────────────────────────── - - def create_repo( - self, - owner: str, - repo_id: str, - name: str, - owner_type: str = "orgs", - repo_type: str = "source", - description: Optional[str] = None, - default_locale: Optional[str] = None, - supported_locales: Optional[list[str]] = None, - public_access: Optional[str] = None, - source_type: Optional[str] = None, - collection_type: Optional[str] = None, - custom_validation_schema: Optional[str] = None, - canonical_url: Optional[str] = None, - extras: Optional[dict] = None, - ) -> dict: - """Create a new repository (source or collection).""" - self._require_auth() - _validate_owner_type(owner_type) - stem = _repo_type_stem(repo_type) - endpoint = f"/{owner_type}/{owner}/{stem}/" - - body: dict[str, Any] = {"id": repo_id, "short_code": repo_id, "name": name} - if description: - body["description"] = description - if default_locale: - body["default_locale"] = default_locale - if supported_locales: - body["supported_locales"] = supported_locales - elif repo_type == "source": - body["supported_locales"] = ["en"] - if public_access: - body["public_access"] = public_access - if source_type and repo_type == "source": - body["source_type"] = source_type - if collection_type and repo_type == "collection": - body["collection_type"] = collection_type - if custom_validation_schema: - body["custom_validation_schema"] = custom_validation_schema - if canonical_url: - body["canonical_url"] = canonical_url - if extras: - body["extras"] = extras - - return self.post(endpoint, json=body) - - def update_repo( - self, - owner: str, - repo: str, - owner_type: str = "orgs", - repo_type: str = "source", - **fields: Any, - ) -> dict: - """Update a repository.""" - self._require_auth() - endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo) - body = {k: v for k, v in fields.items() if v is not None} - return self.patch(endpoint, json=body) - - def create_repo_version( - self, - owner: str, - repo: str, - version_id: str, - owner_type: str = "orgs", - repo_type: str = "source", - description: Optional[str] = None, - released: bool = True, - match_algorithms: Optional[list[str]] = None, - ) -> dict: - """Create a new repository version (snapshot). - - Without match_algorithms, the server decides: a new source version is vectorized - (``llm``) when the source's HEAD is. - """ - self._require_auth() - endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo, suffix="versions/") - body: dict[str, Any] = {"id": version_id, "released": released} - if description: - body["description"] = description - if match_algorithms is not None: - body["match_algorithms"] = match_algorithms - return self.post(endpoint, json=body) - - def update_repo_version( - self, - owner: str, - repo: str, - version_id: str, - owner_type: str = "orgs", - repo_type: str = "source", - **fields: Any, - ) -> dict: - """Update a repository version.""" - self._require_auth() - _validate_owner_type(owner_type) - stem = _repo_type_stem(repo_type) - endpoint = f"/{owner_type}/{owner}/{stem}/{repo}/{version_id}/" - body = {k: v for k, v in fields.items() if v is not None} - return self.patch(endpoint, json=body) - - def create_concept( - self, - owner: str, - source: str, - concept_id: str, - concept_class: str, - names: list[dict], - owner_type: str = "orgs", - datatype: Optional[str] = None, - descriptions: Optional[list[dict]] = None, - external_id: Optional[str] = None, - extras: Optional[dict] = None, - ) -> dict: - """Create a new concept in a source.""" - self._require_auth() - _validate_owner_type(owner_type) - - if not names: - raise ValueError("At least one name is required") - - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/" - body: dict[str, Any] = { - "id": concept_id, - "concept_class": concept_class, - "names": names, - } - if datatype: - body["datatype"] = datatype - if descriptions: - body["descriptions"] = descriptions - if external_id: - body["external_id"] = external_id - if extras: - body["extras"] = extras - - return self.post(endpoint, json=body) - - def update_concept( - self, - owner: str, - source: str, - concept_id: str, - owner_type: str = "orgs", - update_comment: Optional[str] = None, - **fields: Any, - ) -> dict: - """Update an existing concept.""" - self._require_auth() - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/" - body = {k: v for k, v in fields.items() if v is not None} - if update_comment: - body["update_comment"] = update_comment - return self.patch(endpoint, json=body) - - def retire_concept( - self, - owner: str, - source: str, - concept_id: str, - owner_type: str = "orgs", - update_comment: Optional[str] = None, - ) -> dict: - """Retire a concept (set retired=true).""" - return self.update_concept( - owner, source, concept_id, - owner_type=owner_type, - update_comment=update_comment, - retired=True, - ) - - def delete_concept( - self, - owner: str, - source: str, - concept_id: str, - owner_type: str = "orgs", - ) -> dict: - """Hard-delete a concept (?hardDelete=true). Requires staff token.""" - self._require_auth() - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/" - return self.delete(endpoint, params={"hardDelete": "true"}) - - def add_concept_name( - self, - owner: str, - source: str, - concept_id: str, - name: str, - locale: str, - owner_type: str = "orgs", - name_type: Optional[str] = None, - locale_preferred: bool = False, - ) -> dict: - """Add a name/translation to a concept.""" - self._require_auth() - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/names/" - body: dict[str, Any] = {"name": name, "locale": locale} - if name_type: - body["name_type"] = name_type - if locale_preferred: - body["locale_preferred"] = True - return self.post(endpoint, json=body) - - def add_concept_description( - self, - owner: str, - source: str, - concept_id: str, - description: str, - locale: str, - owner_type: str = "orgs", - description_type: Optional[str] = None, - ) -> dict: - """Add a description to a concept.""" - self._require_auth() - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/descriptions/" - body: dict[str, Any] = {"description": description, "locale": locale} - if description_type: - body["description_type"] = description_type - return self.post(endpoint, json=body) - - def set_concept_extra( - self, - owner: str, - source: str, - concept_id: str, - key: str, - value: Any, - owner_type: str = "orgs", - ) -> dict: - """Set a custom attribute on a concept.""" - self._require_auth() - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/extras/{key}/" - return self.put(endpoint, json={key: value}) - - def delete_concept_extra( - self, - owner: str, - source: str, - concept_id: str, - key: str, - owner_type: str = "orgs", - ) -> dict: - """Delete a custom attribute from a concept.""" - self._require_auth() - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/extras/{key}/" - return self.delete(endpoint) - - def create_mapping( - self, - owner: str, - source: str, - map_type: str, - owner_type: str = "orgs", - from_concept_url: Optional[str] = None, - from_source_url: Optional[str] = None, - from_concept_code: Optional[str] = None, - from_concept_name: Optional[str] = None, - to_concept_url: Optional[str] = None, - to_source_url: Optional[str] = None, - to_concept_code: Optional[str] = None, - to_concept_name: Optional[str] = None, - external_id: Optional[str] = None, - extras: Optional[dict] = None, - ) -> dict: - """Create a new mapping between concepts.""" - self._require_auth() - _validate_owner_type(owner_type) - - if not from_concept_url and not (from_source_url and from_concept_code): - raise ValueError("from_concept requires either from_concept_url or (from_source_url + from_concept_code)") - if not to_concept_url and not (to_source_url and to_concept_code): - raise ValueError("to_concept requires either to_concept_url or (to_source_url + to_concept_code)") - - endpoint = f"/{owner_type}/{owner}/sources/{source}/mappings/" - body: dict[str, Any] = {"map_type": map_type} - - if from_concept_url: - body["from_concept_url"] = from_concept_url - else: - body["from_source_url"] = from_source_url - body["from_concept_code"] = from_concept_code - if from_concept_name: - body["from_concept_name"] = from_concept_name - - if to_concept_url: - body["to_concept_url"] = to_concept_url - else: - body["to_source_url"] = to_source_url - body["to_concept_code"] = to_concept_code - if to_concept_name: - body["to_concept_name"] = to_concept_name - - if external_id: - body["external_id"] = external_id - if extras: - body["extras"] = extras - - return self.post(endpoint, json=body) - - def update_mapping( - self, - owner: str, - source: str, - mapping_id: str, - owner_type: str = "orgs", - update_comment: Optional[str] = None, - **fields: Any, - ) -> dict: - """Update an existing mapping.""" - self._require_auth() - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/mappings/{mapping_id}/" - body = {k: v for k, v in fields.items() if v is not None} - if update_comment: - body["update_comment"] = update_comment - return self.patch(endpoint, json=body) - - def retire_mapping( - self, - owner: str, - source: str, - mapping_id: str, - owner_type: str = "orgs", - update_comment: Optional[str] = None, - ) -> dict: - """Retire a mapping.""" - return self.update_mapping( - owner, source, mapping_id, - owner_type=owner_type, - update_comment=update_comment, - retired=True, - ) - - def delete_mapping( - self, - owner: str, - source: str, - mapping_id: str, - owner_type: str = "orgs", - ) -> dict: - """Hard-delete a mapping (?hardDelete=true). Requires staff token.""" - self._require_auth() - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{source}/mappings/{mapping_id}/" - return self.delete(endpoint, params={"hardDelete": "true"}) - - def add_collection_ref( - self, - owner: str, - collection: str, - expressions: list[str], - owner_type: str = "orgs", - cascade: Optional[str] = None, - ) -> dict: - """Add references to a collection.""" - self._require_auth() - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/collections/{collection}/references/" - body: dict[str, Any] = {"data": {"expressions": expressions}} - params: dict[str, Any] = {} - if cascade: - params["cascade"] = cascade - return self.put(endpoint, json=body, params=params or None) - - def remove_collection_ref( - self, - owner: str, - collection: str, - expressions: list[str], - owner_type: str = "orgs", - ) -> dict: - """Remove references from a collection.""" - self._require_auth() - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/collections/{collection}/references/" - # DELETE with body - response = self.client.request( - "DELETE", endpoint, json={"references": expressions} - ) - self._handle_error(response) - if response.status_code == 204: - return {} - return response.json() - - def create_expansion( - self, - owner: str, - collection: str, - version: str, - owner_type: str = "orgs", - **params: Any, - ) -> dict: - """Create/trigger an expansion for a collection version.""" - self._require_auth() - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/collections/{collection}/{version}/expansions/" - body = {k: v for k, v in params.items() if v is not None} - return self.post(endpoint, json=body or None) - - def clone( - self, - owner: str, - dest_source: str, - expressions: list[str], - owner_type: str = "orgs", - map_types: Optional[list[str]] = None, - exclude_map_types: Optional[list[str]] = None, - method: Optional[str] = None, - cascade_levels: Optional[str] = None, - cascade_hierarchy: bool = True, - cascade_mappings: bool = True, - ) -> dict: - """Clone concepts using $clone operation.""" - self._require_auth() - _validate_owner_type(owner_type) - endpoint = f"/{owner_type}/{owner}/sources/{dest_source}/$clone/" - - body: dict[str, Any] = {"expressions": expressions} - params: dict[str, Any] = {} - if map_types: - params["mapTypes"] = ",".join(map_types) - if exclude_map_types: - params["excludeMapTypes"] = ",".join(exclude_map_types) - if method: - params["method"] = method - if cascade_levels: - params["cascadeLevels"] = cascade_levels - if not cascade_hierarchy: - params["cascadeHierarchy"] = "false" - if not cascade_mappings: - params["cascadeMappings"] = "false" - - return self.post(endpoint, json=body, params=params or None) - - def set_repo_extra( - self, - owner: str, - repo: str, - key: str, - value: Any, - owner_type: str = "orgs", - repo_type: str = "source", - ) -> dict: - """Set a custom attribute on a repository.""" - self._require_auth() - endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo, suffix=f"extras/{key}/") - return self.put(endpoint, json={key: value}) - - def delete_repo_extra( - self, - owner: str, - repo: str, - key: str, - owner_type: str = "orgs", - repo_type: str = "source", - ) -> dict: - """Delete a custom attribute from a repository.""" - self._require_auth() - _validate_owner_type(owner_type) - stem = _repo_type_stem(repo_type) - return self.delete(f"/{owner_type}/{owner}/{stem}/{repo}/extras/{key}/") - - def get_repo_extras( - self, - owner: str, - repo: str, - owner_type: str = "orgs", - repo_type: str = "source", - ) -> dict: - """Get custom attributes for a repository.""" - endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo, suffix="extras/") - return self.get(endpoint) - - def add_org_member(self, org: str, username: str) -> dict: - """Add a member to an organization.""" - self._require_auth() - return self.put(f"/orgs/{org}/members/{username}/") - - def remove_org_member(self, org: str, username: str) -> dict: - """Remove a member from an organization.""" - self._require_auth() - return self.delete(f"/orgs/{org}/members/{username}/") - - # ── Task operations ───────────────────────────────────────────── - - def list_tasks( - self, - state: Optional[str] = None, - limit: int = 20, - page: int = 1, - ) -> dict: - """List async tasks for the authenticated user.""" - self._require_auth() - params: dict[str, Any] = {"limit": limit, "page": page} - if state: - params["state"] = state - return self._get_list("/tasks/", params=params) - - def get_task(self, task_id: str) -> dict: - """Get details of a specific task.""" - self._require_auth() - return self.get(f"/tasks/{task_id}/") - - # ── Organization CRUD ─────────────────────────────────────────── - - def create_org( - self, - org_id: str, - name: str, - company: Optional[str] = None, - website: Optional[str] = None, - location: Optional[str] = None, - public_access: str = "View", - extras: Optional[dict] = None, - ) -> dict: - """Create a new organization.""" - self._require_auth() - body: dict[str, Any] = { - "id": org_id, - "name": name, - "public_access": public_access, - } - if company: - body["company"] = company - if website: - body["website"] = website - if location: - body["location"] = location - if extras: - body["extras"] = extras - return self.post("/orgs/", json=body) - - def delete_org(self, org: str) -> dict: - """Delete an organization.""" - self._require_auth() - return self.delete(f"/orgs/{org}/") - - # ── Repository delete ─────────────────────────────────────────── - - def delete_repo( - self, - owner: str, - repo: str, - owner_type: str = "orgs", - repo_type: str = "source", - ) -> dict: - """Delete a repository (source or collection).""" - self._require_auth() - endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo) - return self.delete(endpoint) - - # ── Bulk import ───────────────────────────────────────────────── - - def bulk_import(self, lines: list[dict], update_if_exists: bool = True) -> dict: - """Submit a bulk import (JSON Lines format to /manage/bulkimport/).""" - self._require_auth() - if not lines: - return {"message": "no references"} - import json as _json - body = "\n".join(_json.dumps(item, ensure_ascii=False) for item in lines) + "\n" - params = {} - if update_if_exists: - params["update_if_exists"] = "true" - self._log_request("POST", "/manage/bulkimport/", params) - response = self.client.post( - "/manage/bulkimport/", - content=body.encode("utf-8"), - headers={"Content-Type": "application/json"}, - params=params, - timeout=120.0, - ) - self._handle_error(response) - try: - return response.json() - except Exception: - return {"message": response.text} - - def get_bulk_import_status(self) -> list: - """Get status of bulk import tasks.""" - self._require_auth() - return self.get("/manage/bulkimport/") - - # ── Import (new /importers/bulk-import/ endpoint) ───────────── - - IMPORT_MIME_MAP = { - ".json": "application/json", - ".jsonl": "application/json", - ".csv": "text/csv", - ".zip": "application/zip", - } - - def import_file( - self, - file_path: str, - queue: str | None = None, - update_if_exists: bool = True, - parallel: int | None = None, - ) -> dict: - """Upload a file for bulk import via /importers/bulk-import/. - - Supports .json, .jsonl, .csv, and .zip (OCL export) files. - The API handles format detection and conversion server-side. - """ - import os - - self._require_auth() - - endpoint = "/importers/bulk-import/" - if queue: - endpoint = f"/importers/bulk-import/{queue}/" - - params = {"update_if_exists": "true" if update_if_exists else "false"} - - filename = os.path.basename(file_path) - ext = os.path.splitext(filename)[1].lower() - mime = self.IMPORT_MIME_MAP.get(ext, "application/octet-stream") - - data = {} - if parallel is not None: - data["parallel"] = str(parallel) - - self._log_request("POST", endpoint, params) - with open(file_path, "rb") as f: - response = self.client.post( - endpoint, - files={"file": (filename, f, mime)}, - data=data or None, - params=params, - timeout=120.0, - ) - self._handle_error(response) - try: - return response.json() - except Exception: - return {"message": response.text} - - def import_list(self, queue: str | None = None) -> Any: - """List active/recent bulk imports.""" - self._require_auth() - endpoint = "/importers/bulk-import/" - if queue: - endpoint = f"/importers/bulk-import/{queue}/" - return self.get(endpoint) - - def import_status(self, task_id: str) -> Any: - """Get status of a specific bulk import task.""" - self._require_auth() - return self.get("/importers/bulk-import/", params={"task": task_id}) - - # ── Cascade concept fetching (for pruning) ────────────────────── - - def fetch_cascade_children(self, concept_url: str) -> set[str]: - """Fetch child concept URLs via cascade for a given concept expression.""" - url = concept_url.rstrip("/") + "/cascade/" - params = { - "method": "sourcetoconcepts", - "cascadeHierarchy": "true", - "cascadeMappings": "true", - "cascadeLevels": "*", - "reverse": "false", - "includeRetired": "false", - } - try: - data = self.get(url, params=params) - except (APIError, Exception): - return set() - entries = data.get("entry") or [] - if isinstance(entries, dict): - entries = [entries] - children: set[str] = set() - parent = concept_url.rstrip("/") + "/" - for entry in entries: - if not isinstance(entry, dict) or entry.get("type") != "Concept": - continue - child_url = entry.get("url", "").rstrip("/") + "/" - if child_url and child_url != parent: - children.add(child_url) - return children - - # ── Exports ────────────────────────────────────────────────────── - - def _export_endpoint( - self, owner: str, repo: str, version: str, - owner_type: str, repo_type: str, - ) -> str: - return _build_repo_endpoint( - owner_type, owner, repo_type, repo, version, "export/" - ) - - def export_status( - self, - owner: str, - repo: str, - version: str, - owner_type: str = "orgs", - repo_type: str = "source", - ) -> dict: - """Check export status for a repository version via HEAD request.""" - self._require_auth() - endpoint = self._export_endpoint(owner, repo, version, owner_type, repo_type) - response = self.request("HEAD", endpoint, timeout=45.0) - code = response.status_code - if code in (200, 302): - filename = "" - cd = response.headers.get("content-disposition", "") - if "filename=" in cd: - filename = cd.split("filename=", 1)[1].strip().strip('"') - url = response.headers.get("location", "") - return {"status": "ready", "status_code": code, "filename": filename, "url": url} - elif code == 204: - return {"status": "not_found", "status_code": 204} - elif code == 208: - return {"status": "processing", "status_code": 208} - elif code == 404: - raise APIError("Repository version not found", status_code=404) - else: - self._handle_error(response) - return {"status": "unknown", "status_code": code} - - def export_create( - self, - owner: str, - repo: str, - version: str, - owner_type: str = "orgs", - repo_type: str = "source", - ) -> dict: - """Trigger export creation for a repository version.""" - self._require_auth() - endpoint = self._export_endpoint(owner, repo, version, owner_type, repo_type) - response = self.request("POST", endpoint, timeout=45.0) - code = response.status_code - if code == 202: - return {"status": "accepted", "status_code": 202, - "message": "Export creation started"} - elif code in (200, 303): - return {"status": "already_exists", "status_code": code, - "message": "Export already exists"} - elif code == 409: - return {"status": "conflict", "status_code": 409, - "message": "Export is already being processed"} - else: - self._handle_error(response) - return {"status": "unknown", "status_code": code} - - def export_delete( - self, - owner: str, - repo: str, - version: str, - owner_type: str = "orgs", - repo_type: str = "source", - ) -> dict: - """Delete an export for a repository version.""" - self._require_auth() - endpoint = self._export_endpoint(owner, repo, version, owner_type, repo_type) - response = self.request("DELETE", endpoint, timeout=30.0) - if response.status_code == 204: - return {"status": "deleted", "status_code": 204} - elif response.status_code == 404: - raise APIError("Export not found", status_code=404) - else: - self._handle_error(response) - return {"status": "unknown", "status_code": response.status_code} - - def export_download( - self, - owner: str, - repo: str, - version: str, - owner_type: str = "orgs", - repo_type: str = "source", - ) -> httpx.Response: - """Download an export file. Returns the streaming response. - - The response has Content-Type: application/zip and a - Content-Disposition header with the filename. - Caller should consume and close the response. - """ - self._require_auth() - endpoint = self._export_endpoint(owner, repo, version, owner_type, repo_type) - response = self.request("GET", endpoint, timeout=300.0) - code = response.status_code - if code == 200: - return response - elif code == 302: - # API returns 302 redirect to S3 signed URL — follow it - redirect_url = response.headers.get("location", "") - if not redirect_url: - raise APIError("Export redirect missing Location header", status_code=302) - redirect_response = httpx.get(redirect_url, timeout=300.0) - redirect_response.raise_for_status() - return redirect_response - elif code == 204: - raise APIError( - "Export does not exist. Use 'ocl repo export create' first.", - status_code=204, - ) - elif code == 208: - raise APIError( - "Export is still being processed. Try again shortly.", - status_code=208, - ) - elif code == 404: - raise APIError("Repository version not found", status_code=404) - else: - self._handle_error(response) - raise APIError(f"Unexpected status {code}", status_code=code) +__all__ = ["APIError", "OCLAPIClient"] diff --git a/src/openconceptlab/__init__.py b/src/openconceptlab/__init__.py new file mode 100644 index 0000000..c969cb7 --- /dev/null +++ b/src/openconceptlab/__init__.py @@ -0,0 +1,12 @@ +"""Open Concept Lab API client, shared by the `ocl` command and other OCL tools.""" + +from importlib.metadata import PackageNotFoundError, version + +try: + __version__ = version("openconceptlab") +except PackageNotFoundError: # a source tree that isn't installed + __version__ = "0.0.0" + +from openconceptlab.client import APIError, OCLAPIClient # noqa: E402 + +__all__ = ["APIError", "OCLAPIClient", "__version__"] diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py new file mode 100644 index 0000000..809b331 --- /dev/null +++ b/src/openconceptlab/client.py @@ -0,0 +1,1663 @@ +"""Synchronous client for the Open Concept Lab API (httpx, with retries). + +The open core shared by the `ocl` command and other OCL tools. +""" + +import platform +import sys +from typing import Any, Optional + +import httpx +from tenacity import ( + retry, + retry_if_exception_type, + stop_after_attempt, + wait_exponential, +) + +from openconceptlab import __version__ + + +class APIError(Exception): + """Structured API error with status code and detail.""" + + def __init__(self, message: str, status_code: int = 0, detail: str = ""): + super().__init__(message) + self.status_code = status_code + self.detail = detail + + def to_dict(self) -> dict: + d = {"error": str(self), "status_code": self.status_code} + if self.detail: + d["detail"] = self.detail + return d + + +def _repo_type_stem(repo_type: str) -> str: + """Convert 'source'/'collection' to plural URL segment.""" + if repo_type == "source": + return "sources" + elif repo_type == "collection": + return "collections" + else: + raise ValueError(f"repo_type must be 'source' or 'collection', got '{repo_type}'") + + +def _validate_owner_type(owner_type: str) -> None: + if owner_type not in ("users", "orgs"): + raise ValueError(f"owner_type must be 'users' or 'orgs', got '{owner_type}'") + + +def _build_repo_endpoint( + owner_type: str, + owner: str, + repo_type: str, + repo: str, + repo_version: Optional[str] = None, + suffix: str = "", +) -> str: + """Build a repository-scoped endpoint URL.""" + _validate_owner_type(owner_type) + stem = _repo_type_stem(repo_type) + base = f"/{owner_type}/{owner}/{stem}/{repo}" + if repo_version: + base = f"{base}/{repo_version}" + return f"{base}/{suffix}" if suffix else f"{base}/" + + +class OCLAPIClient: + """Synchronous client for the OCL API.""" + + DEFAULT_BASE_URL = "https://api.openconceptlab.org" + + def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None): + self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/") + self.token = token + self.debug = False + self.show_request = False + self._show_request_server_printed = False + + headers = { + "Accept": "application/json", + "X-OCL-CLIENT": f"ocl-cli/{__version__}", + "User-Agent": f"ocl-cli/{__version__} (httpx/{httpx.__version__}; python/{platform.python_version()})", + } + if self.token: + headers["Authorization"] = f"Token {self.token}" + + self.client = httpx.Client( + base_url=self.base_url, + headers=headers, + timeout=30.0, + limits=httpx.Limits(max_keepalive_connections=10, max_connections=20), + ) + + def close(self): + self.client.close() + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + + def _log_request(self, method: str, endpoint: str, params: dict | None = None, + body: Any = None): + if self.debug: + url = f"{self.base_url}{endpoint}" + if params: + from urllib.parse import urlencode + url = f"{url}?{urlencode(params, doseq=True)}" + print(f" {method} {url}", file=sys.stderr) + if body is not None: + import json as _json + print(f" Body: {_json.dumps(body, default=str)}", file=sys.stderr) + if self.show_request: + if not self._show_request_server_printed: + print(f"Server: {self.base_url}", file=sys.stderr) + self._show_request_server_printed = True + url = endpoint + if params: + from urllib.parse import urlencode + url = f"{url}?{urlencode(params, doseq=True)}" + print(f" → {method} {url}", file=sys.stderr) + + def _handle_error(self, response: httpx.Response) -> None: + """Convert HTTP errors to APIError.""" + if response.status_code == 429: + retry_after = response.headers.get("Retry-After", "60") + raise APIError( + f"Rate limit exceeded. Retry after {retry_after}s", + status_code=429, + ) + if response.status_code >= 400: + try: + detail = response.json() + except Exception: + detail = response.text + raise APIError( + f"HTTP {response.status_code}", + status_code=response.status_code, + detail=str(detail), + ) + + @retry( + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=4, max=60), + retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), + ) + def get(self, endpoint: str, params: Optional[dict] = None, timeout: Optional[float] = None) -> Any: + self._log_request("GET", endpoint, params) + kwargs: dict[str, Any] = {"params": params} + if timeout is not None: + kwargs["timeout"] = timeout + response = self.client.get(endpoint, **kwargs) + self._handle_error(response) + return response.json() + + @retry( + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=4, max=60), + retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), + ) + def post(self, endpoint: str, json: Optional[dict] = None, params: Optional[dict] = None) -> Any: + self._log_request("POST", endpoint, params, body=json) + response = self.client.post(endpoint, json=json, params=params) + self._handle_error(response) + return response.json() + + @retry( + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=4, max=60), + retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), + ) + def put(self, endpoint: str, json: Optional[dict] = None, params: Optional[dict] = None) -> Any: + self._log_request("PUT", endpoint, params, body=json) + response = self.client.put(endpoint, json=json, params=params) + self._handle_error(response) + return response.json() + + @retry( + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=4, max=60), + retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), + ) + def patch(self, endpoint: str, json: Optional[dict] = None, params: Optional[dict] = None) -> Any: + self._log_request("PATCH", endpoint, params, body=json) + response = self.client.patch(endpoint, json=json, params=params) + self._handle_error(response) + return response.json() + + @retry( + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=4, max=60), + retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), + ) + def delete(self, endpoint: str, params: Optional[dict] = None) -> Any: + self._log_request("DELETE", endpoint, params) + response = self.client.delete(endpoint, params=params) + self._handle_error(response) + if response.status_code == 204: + return {} + return response.json() + + def request( + self, + method: str, + endpoint: str, + params: Optional[dict] = None, + json: Optional[dict] = None, + follow_redirects: bool = False, + timeout: Optional[float] = None, + ) -> httpx.Response: + """Low-level request returning the raw httpx.Response. + + Unlike get/post/delete, this does NOT call _handle_error(). + Callers are responsible for interpreting status codes. + Used for APIs with non-standard status code semantics (e.g. exports). + """ + self._log_request(method, endpoint, params, body=json) + kwargs: dict[str, Any] = {"params": params, "follow_redirects": follow_redirects} + if json is not None: + kwargs["json"] = json + if timeout is not None: + kwargs["timeout"] = timeout + return self.client.request(method, endpoint, **kwargs) + + # ── Helpers ────────────────────────────────────────────────────── + + def _normalize(self, response: Any, headers: Optional[dict] = None) -> dict: + """Normalize list responses to {count, results} format. + + When ``headers`` are provided (from _get_list), ``num_found`` is used + as the authoritative total count for pagination. + """ + num_found = int(headers.get("num_found", 0)) if headers else 0 + if isinstance(response, list): + return {"count": num_found or len(response), "results": response} + if isinstance(response, dict) and num_found: + response["count"] = num_found + return response + + @retry( + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=4, max=60), + retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), + ) + def _get_list(self, endpoint: str, params: Optional[dict] = None) -> dict: + """GET a list endpoint, preserving pagination headers. + + The OCL API returns lists with pagination metadata in response + headers (num_found, num_returned, page_number). The standard + get() discards headers, so this method captures them and feeds + them to _normalize() for correct total counts. + """ + self._log_request("GET", endpoint, params) + response = self.client.get(endpoint, params=params) + self._handle_error(response) + return self._normalize(response.json(), headers=dict(response.headers)) + + def _require_auth(self) -> None: + if not self.token: + raise APIError("Authentication required. Run 'ocl login' first.", status_code=401) + + # ── High-level API methods ─────────────────────────────────────── + + def get_user(self) -> dict: + """Get current authenticated user info.""" + self._require_auth() + return self.get("/user/") + + # ── Organization operations ──────────────────────────────────── + + def list_orgs( + self, + query: Optional[str] = None, + verbose: bool = False, + limit: int = 25, + page: int = 1, + ) -> dict: + """List/search organizations.""" + params: dict[str, Any] = {"limit": limit, "page": page} + if query: + params["q"] = query + if verbose: + params["verbose"] = True + return self._get_list("/orgs/", params=params) + + def get_org(self, org: str) -> dict: + """Get a specific organization.""" + return self.get(f"/orgs/{org}/") + + def list_org_repos( + self, + org: str, + repo_type: str = "all", + limit: int = 20, + page: int = 1, + ) -> dict: + """List repositories owned by an organization.""" + params: dict[str, Any] = {"limit": limit, "page": page} + if repo_type != "all": + params["repoType"] = "Source" if repo_type == "source" else "Collection" + return self._get_list(f"/orgs/{org}/repos/", params=params) + + # ── User operations ────────────────────────────────────────── + + def list_users( + self, + query: Optional[str] = None, + verbose: bool = False, + limit: int = 25, + page: int = 1, + ) -> dict: + """List/search users.""" + params: dict[str, Any] = {"limit": limit, "page": page} + if query: + params["q"] = query + if verbose: + params["verbose"] = True + return self._get_list("/users/", params=params) + + def get_user_detail(self, username: str) -> dict: + """Get a specific user by username.""" + return self.get(f"/users/{username}/") + + def list_user_repos( + self, + username: str, + repo_type: str = "all", + limit: int = 20, + page: int = 1, + ) -> dict: + """List repositories owned by a user.""" + params: dict[str, Any] = {"limit": limit, "page": page} + if repo_type != "all": + params["repoType"] = "Source" if repo_type == "source" else "Collection" + return self._get_list(f"/users/{username}/repos/", params=params) + + def list_user_orgs(self, username: str, limit: int = 100) -> dict: + """List organizations a user belongs to.""" + return self._get_list(f"/users/{username}/orgs/", params={"limit": limit, "verbose": "true"}) + + def get_org_members(self, org: str, limit: int = 100) -> dict: + """List members of an organization.""" + return self._get_list(f"/orgs/{org}/members/", params={"limit": limit}) + + def search_repos( + self, + query: Optional[str] = None, + owner: Optional[str] = None, + owner_type: str = "all", + repo_type: str = "all", + custom_validation_schema: Optional[str] = None, + updated_since: Optional[str] = None, + all_versions: bool = False, + verbose: bool = False, + limit: int = 20, + page: int = 1, + ) -> dict: + """Search for repositories (sources and collections).""" + params: dict[str, Any] = {"limit": limit, "page": page} + if verbose: + params["verbose"] = True + if query: + params["q"] = query + if repo_type != "all": + params["repoType"] = "Source" if repo_type == "source" else "Collection" + if custom_validation_schema: + params["customValidationSchema"] = custom_validation_schema + if updated_since: + params["updatedSince"] = updated_since + if all_versions: + params["allVersions"] = "true" + + # Use scoped endpoint when owner is specified + if owner and owner_type and owner_type != "all": + ot = owner_type # "orgs" or "users" + endpoint = f"/{ot}/{owner}/repos/" + elif owner: + # Default to orgs scope + endpoint = f"/orgs/{owner}/repos/" + else: + # Global search — add owner/ownerType as query params + endpoint = "/repos/" + if owner_type != "all": + params["ownerType"] = "Organization" if owner_type == "orgs" else "User" + + return self._get_list(endpoint, params=params) + + def get_repo( + self, + owner: str, + repo: str, + owner_type: str = "orgs", + repo_type: str = "source", + repo_version: Optional[str] = None, + include_summary: bool = False, + ) -> dict: + """Get a specific repository, optionally at a version.""" + endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo, repo_version) + params = {} + if include_summary: + params["includeSummary"] = "true" + return self.get(endpoint, params=params or None) + + def get_repo_versions( + self, + owner: str, + repo: str, + owner_type: str = "orgs", + repo_type: str = "source", + released: Optional[bool] = None, + processing: Optional[bool] = None, + updated_since: Optional[str] = None, + limit: int = 20, + page: int = 1, + ) -> dict: + """List versions for a repository.""" + endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo, suffix="versions/") + params: dict[str, Any] = {"limit": limit, "page": page} + if released is not None: + params["released"] = str(released).lower() + if processing is not None: + params["processing"] = str(processing).lower() + if updated_since: + params["updatedSince"] = updated_since + return self._get_list(endpoint, params=params) + + def search_concepts( + self, + query: Optional[str] = None, + owner: Optional[str] = None, + owner_type: Optional[str] = None, + repo_type: Optional[str] = None, + repo: Optional[str] = None, + repo_version: Optional[str] = None, + concept_class: Optional[str] = None, + datatype: Optional[str] = None, + locale: Optional[str] = None, + include_retired: bool = False, + include_mappings: bool = False, + include_inverse_mappings: bool = False, + updated_since: Optional[str] = None, + sort: Optional[str] = None, + verbose: bool = False, + limit: int = 20, + page: int = 1, + ) -> dict: + """Search for concepts globally or within a repository.""" + params: dict[str, Any] = {"limit": limit, "page": page} + if verbose: + params["verbose"] = True + if query: + params["q"] = query + if concept_class: + params["conceptClass"] = concept_class + if datatype: + params["datatype"] = datatype + if locale: + params["locale"] = locale + if include_retired: + params["includeRetired"] = "true" + if include_mappings: + params["includeMappings"] = "true" + if include_inverse_mappings: + params["includeInverseMappings"] = "true" + if updated_since: + params["updatedSince"] = updated_since + if sort: + params["sortAsc"] = sort if not sort.startswith("-") else None + if sort.startswith("-"): + params["sortDesc"] = sort[1:] + # Determine endpoint scope — use narrowest available + if all([owner_type, owner, repo_type, repo]): + endpoint = _build_repo_endpoint( + owner_type, owner, repo_type, repo, repo_version, suffix="concepts/" + ) + elif owner and repo: + ot = owner_type or "orgs" + rt = repo_type or "source" + endpoint = _build_repo_endpoint(ot, owner, rt, repo, repo_version, suffix="concepts/") + else: + # Global search — add owner/ownerType as query params + endpoint = "/concepts/" + if owner: + params["owner"] = owner + if owner_type and owner_type not in ("all",): + params["ownerType"] = "Organization" if owner_type == "orgs" else "User" + + return self._get_list(endpoint, params=params) + + def get_concept( + self, + owner: str, + source: str, + concept_id: str, + owner_type: str = "orgs", + repo_version: Optional[str] = None, + concept_version: Optional[str] = None, + include_mappings: bool = False, + include_inverse_mappings: bool = False, + verbose: bool = False, + ) -> dict: + """Get a single concept.""" + _validate_owner_type(owner_type) + stem = _repo_type_stem("source") + base = f"/{owner_type}/{owner}/{stem}/{source}" + if repo_version: + base = f"{base}/{repo_version}" + endpoint = f"{base}/concepts/{concept_id}/" + if concept_version: + endpoint = f"{base}/concepts/{concept_id}/{concept_version}/" + + params: dict[str, Any] = {} + if include_mappings: + params["includeMappings"] = "true" + if include_inverse_mappings: + params["includeInverseMappings"] = "true" + if verbose: + params["verbose"] = "true" + + return self.get(endpoint, params=params or None) + + def get_concept_versions( + self, + owner: str, + source: str, + concept_id: str, + owner_type: str = "orgs", + limit: int = 20, + page: int = 1, + ) -> dict: + """List version history for a concept.""" + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/versions/" + return self._get_list(endpoint, params={"limit": limit, "page": page}) + + def get_concept_names( + self, + owner: str, + source: str, + concept_id: str, + owner_type: str = "orgs", + ) -> dict: + """List names for a concept.""" + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/names/" + return self._get_list(endpoint) + + def get_concept_descriptions( + self, + owner: str, + source: str, + concept_id: str, + owner_type: str = "orgs", + ) -> dict: + """List descriptions for a concept.""" + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/descriptions/" + return self._get_list(endpoint) + + def get_concept_extras( + self, + owner: str, + source: str, + concept_id: str, + owner_type: str = "orgs", + ) -> dict: + """Get extras (custom attributes) for a concept.""" + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/extras/" + return self.get(endpoint) + + def search_mappings( + self, + query: Optional[str] = None, + owner: Optional[str] = None, + owner_type: Optional[str] = None, + repo_type: Optional[str] = None, + repo: Optional[str] = None, + repo_version: Optional[str] = None, + map_type: Optional[str] = None, + from_source: Optional[str] = None, + from_concept: Optional[str] = None, + from_concept_owner: Optional[str] = None, + to_source: Optional[str] = None, + to_concept: Optional[str] = None, + to_concept_owner: Optional[str] = None, + include_retired: bool = False, + updated_since: Optional[str] = None, + sort: Optional[str] = None, + verbose: bool = False, + limit: int = 20, + page: int = 1, + ) -> dict: + """Search for mappings globally or within a repository.""" + params: dict[str, Any] = {"limit": limit, "page": page} + if verbose: + params["verbose"] = True + if query: + params["q"] = query + if map_type: + params["mapType"] = map_type + if from_source: + params["fromConceptSource"] = from_source + if from_concept: + params["fromConcept"] = from_concept + if from_concept_owner: + params["fromConceptOwner"] = from_concept_owner + if to_source: + params["toConceptSource"] = to_source + if to_concept: + params["toConcept"] = to_concept + if to_concept_owner: + params["toConceptOwner"] = to_concept_owner + if include_retired: + params["includeRetired"] = "true" + if updated_since: + params["updatedSince"] = updated_since + if sort: + params["sortAsc"] = sort if not sort.startswith("-") else None + if sort.startswith("-"): + params["sortDesc"] = sort[1:] + # Determine endpoint scope — use narrowest available + if all([owner_type, owner, repo_type, repo]): + endpoint = _build_repo_endpoint( + owner_type, owner, repo_type, repo, repo_version, suffix="mappings/" + ) + elif owner and repo: + ot = owner_type or "orgs" + rt = repo_type or "source" + endpoint = _build_repo_endpoint(ot, owner, rt, repo, repo_version, suffix="mappings/") + else: + # Global search — add owner/ownerType as query params + endpoint = "/mappings/" + if owner: + params["owner"] = owner + if owner_type and owner_type not in ("all",): + params["ownerType"] = "Organization" if owner_type == "orgs" else "User" + + return self._get_list(endpoint, params=params) + + def get_mapping( + self, + owner: str, + source: str, + mapping_id: str, + owner_type: str = "orgs", + repo_version: Optional[str] = None, + ) -> dict: + """Get a single mapping.""" + _validate_owner_type(owner_type) + base = f"/{owner_type}/{owner}/sources/{source}" + if repo_version: + base = f"{base}/{repo_version}" + return self.get(f"{base}/mappings/{mapping_id}/") + + def get_mapping_versions( + self, + owner: str, + source: str, + mapping_id: str, + owner_type: str = "orgs", + limit: int = 20, + page: int = 1, + ) -> dict: + """List version history for a mapping.""" + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/mappings/{mapping_id}/versions/" + return self._get_list(endpoint, params={"limit": limit, "page": page}) + + def cascade( + self, + owner: str, + repo: str, + concept_id: str, + owner_type: str = "orgs", + repo_type: str = "source", + repo_version: Optional[str] = None, + map_types: Optional[list[str]] = None, + exclude_map_types: Optional[list[str]] = None, + return_map_types: Optional[list[str]] = None, + method: str = "sourcetoconcepts", + cascade_hierarchy: bool = True, + cascade_mappings: bool = True, + cascade_levels: str = "*", + reverse: bool = False, + view: str = "hierarchy", + verbose: bool = False, + omit_if_exists_in: Optional[str] = None, + equivalency_map_type: Optional[str] = None, + ) -> dict: + """Execute $cascade operation on a concept.""" + endpoint = _build_repo_endpoint( + owner_type, owner, repo_type, repo, repo_version, + suffix=f"concepts/{concept_id}/$cascade/" + ) + + params: dict[str, Any] = {} + if verbose: + params["verbose"] = True + if map_types: + params["mapTypes"] = ",".join(map_types) + if exclude_map_types: + params["excludeMapTypes"] = ",".join(exclude_map_types) + if return_map_types: + params["returnMapTypes"] = ",".join(return_map_types) + if method != "sourcetoconcepts": + params["method"] = method + if not cascade_hierarchy: + params["cascadeHierarchy"] = "false" + if not cascade_mappings: + params["cascadeMappings"] = "false" + if cascade_levels != "*": + params["cascadeLevels"] = cascade_levels + if reverse: + params["reverse"] = "true" + if view != "hierarchy": + params["view"] = view + else: + params["view"] = "hierarchy" + if omit_if_exists_in: + params["omitIfExistsIn"] = omit_if_exists_in + if equivalency_map_type: + params["equivalencyMapType"] = equivalency_map_type + + result = self.get(endpoint, params=params or None, timeout=120.0) + if isinstance(result, list): + return {"resourceType": "Bundle", "type": "searchset", "entry": result} + return result + + def match_concepts( + self, + terms: list[str], + target_repo_url: str, + concept_class: Optional[str] = None, + datatype: Optional[str] = None, + limit: int = 5, + include_retired: bool = False, + include_mappings: bool = False, + semantic: bool = True, + verbose: bool = False, + ) -> dict: + """Match concepts using the $match endpoint.""" + rows = [{"name": term} for term in terms] + body: dict[str, Any] = { + "rows": rows, + "target_repo_url": target_repo_url, + } + filters: dict[str, Any] = {} + if concept_class: + filters["concept_class"] = concept_class + if datatype: + filters["datatype"] = datatype + if filters: + body["filter"] = filters + params: dict[str, Any] = { + "includeSearchMeta": True, + "semantic": semantic, + "bestMatch": True, + } + if limit: + params["limit"] = limit + if include_retired: + params["includeRetired"] = True + if include_mappings: + params["includeMappings"] = True + if verbose: + params["verbose"] = True + + result = self.post("/concepts/$match/", params=params, json=body) + return self._normalize(result) + + def list_expansions( + self, + owner: str, + collection: str, + collection_version: str, + owner_type: str = "orgs", + ) -> dict: + """List expansions for a collection version.""" + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/collections/{collection}/{collection_version}/expansions/" + return self._get_list(endpoint, params={"verbose": True}) + + def get_expansion( + self, + owner: str, + collection: str, + owner_type: str = "orgs", + collection_version: Optional[str] = None, + expansion_id: Optional[str] = None, + ) -> dict: + """Get a specific or default expansion for a collection.""" + _validate_owner_type(owner_type) + + if expansion_id: + if collection_version: + endpoint = f"/{owner_type}/{owner}/collections/{collection}/{collection_version}/expansions/{expansion_id}/" + else: + endpoint = f"/{owner_type}/{owner}/collections/{collection}/expansions/{expansion_id}/" + return self.get(endpoint, params={"verbose": True}) + + # Look up default expansion from collection version + if collection_version: + col_endpoint = f"/{owner_type}/{owner}/collections/{collection}/{collection_version}/" + else: + col_endpoint = f"/{owner_type}/{owner}/collections/{collection}/" + + col_data = self.get(col_endpoint) + expansion_url = col_data.get("expansion_url") + if expansion_url: + return self.get(expansion_url, params={"verbose": True}) + return { + "message": "No default expansion configured for this collection version", + "expansion_url": None, + } + + def list_collection_refs( + self, + owner: str, + collection: str, + owner_type: str = "orgs", + collection_version: Optional[str] = None, + limit: int = 20, + page: int = 1, + ) -> dict: + """List references in a collection.""" + _validate_owner_type(owner_type) + base = f"/{owner_type}/{owner}/collections/{collection}" + if collection_version: + base = f"{base}/{collection_version}" + endpoint = f"{base}/references/" + return self._get_list(endpoint, params={"limit": limit, "page": page}) + + def resolve_reference(self, *expressions: str, namespace: Optional[str] = None) -> dict: + """Resolve references using $resolveReference.""" + params: dict[str, Any] = {} + if namespace: + params["namespace"] = namespace + body = {"references": list(expressions)} + return self.post("/$resolveReference/", json=body, params=params or None) + + # ── Write operations ───────────────────────────────────────────── + + def create_repo( + self, + owner: str, + repo_id: str, + name: str, + owner_type: str = "orgs", + repo_type: str = "source", + description: Optional[str] = None, + default_locale: Optional[str] = None, + supported_locales: Optional[list[str]] = None, + public_access: Optional[str] = None, + source_type: Optional[str] = None, + collection_type: Optional[str] = None, + custom_validation_schema: Optional[str] = None, + canonical_url: Optional[str] = None, + extras: Optional[dict] = None, + ) -> dict: + """Create a new repository (source or collection).""" + self._require_auth() + _validate_owner_type(owner_type) + stem = _repo_type_stem(repo_type) + endpoint = f"/{owner_type}/{owner}/{stem}/" + + body: dict[str, Any] = {"id": repo_id, "short_code": repo_id, "name": name} + if description: + body["description"] = description + if default_locale: + body["default_locale"] = default_locale + if supported_locales: + body["supported_locales"] = supported_locales + elif repo_type == "source": + body["supported_locales"] = ["en"] + if public_access: + body["public_access"] = public_access + if source_type and repo_type == "source": + body["source_type"] = source_type + if collection_type and repo_type == "collection": + body["collection_type"] = collection_type + if custom_validation_schema: + body["custom_validation_schema"] = custom_validation_schema + if canonical_url: + body["canonical_url"] = canonical_url + if extras: + body["extras"] = extras + + return self.post(endpoint, json=body) + + def update_repo( + self, + owner: str, + repo: str, + owner_type: str = "orgs", + repo_type: str = "source", + **fields: Any, + ) -> dict: + """Update a repository.""" + self._require_auth() + endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo) + body = {k: v for k, v in fields.items() if v is not None} + return self.patch(endpoint, json=body) + + def create_repo_version( + self, + owner: str, + repo: str, + version_id: str, + owner_type: str = "orgs", + repo_type: str = "source", + description: Optional[str] = None, + released: bool = True, + match_algorithms: Optional[list[str]] = None, + ) -> dict: + """Create a new repository version (snapshot). + + Without match_algorithms, the server decides: a new source version is vectorized + (``llm``) when the source's HEAD is. + """ + self._require_auth() + endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo, suffix="versions/") + body: dict[str, Any] = {"id": version_id, "released": released} + if description: + body["description"] = description + if match_algorithms is not None: + body["match_algorithms"] = match_algorithms + return self.post(endpoint, json=body) + + def update_repo_version( + self, + owner: str, + repo: str, + version_id: str, + owner_type: str = "orgs", + repo_type: str = "source", + **fields: Any, + ) -> dict: + """Update a repository version.""" + self._require_auth() + _validate_owner_type(owner_type) + stem = _repo_type_stem(repo_type) + endpoint = f"/{owner_type}/{owner}/{stem}/{repo}/{version_id}/" + body = {k: v for k, v in fields.items() if v is not None} + return self.patch(endpoint, json=body) + + def create_concept( + self, + owner: str, + source: str, + concept_id: str, + concept_class: str, + names: list[dict], + owner_type: str = "orgs", + datatype: Optional[str] = None, + descriptions: Optional[list[dict]] = None, + external_id: Optional[str] = None, + extras: Optional[dict] = None, + ) -> dict: + """Create a new concept in a source.""" + self._require_auth() + _validate_owner_type(owner_type) + + if not names: + raise ValueError("At least one name is required") + + endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/" + body: dict[str, Any] = { + "id": concept_id, + "concept_class": concept_class, + "names": names, + } + if datatype: + body["datatype"] = datatype + if descriptions: + body["descriptions"] = descriptions + if external_id: + body["external_id"] = external_id + if extras: + body["extras"] = extras + + return self.post(endpoint, json=body) + + def update_concept( + self, + owner: str, + source: str, + concept_id: str, + owner_type: str = "orgs", + update_comment: Optional[str] = None, + **fields: Any, + ) -> dict: + """Update an existing concept.""" + self._require_auth() + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/" + body = {k: v for k, v in fields.items() if v is not None} + if update_comment: + body["update_comment"] = update_comment + return self.patch(endpoint, json=body) + + def retire_concept( + self, + owner: str, + source: str, + concept_id: str, + owner_type: str = "orgs", + update_comment: Optional[str] = None, + ) -> dict: + """Retire a concept (set retired=true).""" + return self.update_concept( + owner, source, concept_id, + owner_type=owner_type, + update_comment=update_comment, + retired=True, + ) + + def delete_concept( + self, + owner: str, + source: str, + concept_id: str, + owner_type: str = "orgs", + ) -> dict: + """Hard-delete a concept (?hardDelete=true). Requires staff token.""" + self._require_auth() + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/" + return self.delete(endpoint, params={"hardDelete": "true"}) + + def add_concept_name( + self, + owner: str, + source: str, + concept_id: str, + name: str, + locale: str, + owner_type: str = "orgs", + name_type: Optional[str] = None, + locale_preferred: bool = False, + ) -> dict: + """Add a name/translation to a concept.""" + self._require_auth() + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/names/" + body: dict[str, Any] = {"name": name, "locale": locale} + if name_type: + body["name_type"] = name_type + if locale_preferred: + body["locale_preferred"] = True + return self.post(endpoint, json=body) + + def add_concept_description( + self, + owner: str, + source: str, + concept_id: str, + description: str, + locale: str, + owner_type: str = "orgs", + description_type: Optional[str] = None, + ) -> dict: + """Add a description to a concept.""" + self._require_auth() + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/descriptions/" + body: dict[str, Any] = {"description": description, "locale": locale} + if description_type: + body["description_type"] = description_type + return self.post(endpoint, json=body) + + def set_concept_extra( + self, + owner: str, + source: str, + concept_id: str, + key: str, + value: Any, + owner_type: str = "orgs", + ) -> dict: + """Set a custom attribute on a concept.""" + self._require_auth() + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/extras/{key}/" + return self.put(endpoint, json={key: value}) + + def delete_concept_extra( + self, + owner: str, + source: str, + concept_id: str, + key: str, + owner_type: str = "orgs", + ) -> dict: + """Delete a custom attribute from a concept.""" + self._require_auth() + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/concepts/{concept_id}/extras/{key}/" + return self.delete(endpoint) + + def create_mapping( + self, + owner: str, + source: str, + map_type: str, + owner_type: str = "orgs", + from_concept_url: Optional[str] = None, + from_source_url: Optional[str] = None, + from_concept_code: Optional[str] = None, + from_concept_name: Optional[str] = None, + to_concept_url: Optional[str] = None, + to_source_url: Optional[str] = None, + to_concept_code: Optional[str] = None, + to_concept_name: Optional[str] = None, + external_id: Optional[str] = None, + extras: Optional[dict] = None, + ) -> dict: + """Create a new mapping between concepts.""" + self._require_auth() + _validate_owner_type(owner_type) + + if not from_concept_url and not (from_source_url and from_concept_code): + raise ValueError("from_concept requires either from_concept_url or (from_source_url + from_concept_code)") + if not to_concept_url and not (to_source_url and to_concept_code): + raise ValueError("to_concept requires either to_concept_url or (to_source_url + to_concept_code)") + + endpoint = f"/{owner_type}/{owner}/sources/{source}/mappings/" + body: dict[str, Any] = {"map_type": map_type} + + if from_concept_url: + body["from_concept_url"] = from_concept_url + else: + body["from_source_url"] = from_source_url + body["from_concept_code"] = from_concept_code + if from_concept_name: + body["from_concept_name"] = from_concept_name + + if to_concept_url: + body["to_concept_url"] = to_concept_url + else: + body["to_source_url"] = to_source_url + body["to_concept_code"] = to_concept_code + if to_concept_name: + body["to_concept_name"] = to_concept_name + + if external_id: + body["external_id"] = external_id + if extras: + body["extras"] = extras + + return self.post(endpoint, json=body) + + def update_mapping( + self, + owner: str, + source: str, + mapping_id: str, + owner_type: str = "orgs", + update_comment: Optional[str] = None, + **fields: Any, + ) -> dict: + """Update an existing mapping.""" + self._require_auth() + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/mappings/{mapping_id}/" + body = {k: v for k, v in fields.items() if v is not None} + if update_comment: + body["update_comment"] = update_comment + return self.patch(endpoint, json=body) + + def retire_mapping( + self, + owner: str, + source: str, + mapping_id: str, + owner_type: str = "orgs", + update_comment: Optional[str] = None, + ) -> dict: + """Retire a mapping.""" + return self.update_mapping( + owner, source, mapping_id, + owner_type=owner_type, + update_comment=update_comment, + retired=True, + ) + + def delete_mapping( + self, + owner: str, + source: str, + mapping_id: str, + owner_type: str = "orgs", + ) -> dict: + """Hard-delete a mapping (?hardDelete=true). Requires staff token.""" + self._require_auth() + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{source}/mappings/{mapping_id}/" + return self.delete(endpoint, params={"hardDelete": "true"}) + + def add_collection_ref( + self, + owner: str, + collection: str, + expressions: list[str], + owner_type: str = "orgs", + cascade: Optional[str] = None, + ) -> dict: + """Add references to a collection.""" + self._require_auth() + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/collections/{collection}/references/" + body: dict[str, Any] = {"data": {"expressions": expressions}} + params: dict[str, Any] = {} + if cascade: + params["cascade"] = cascade + return self.put(endpoint, json=body, params=params or None) + + def remove_collection_ref( + self, + owner: str, + collection: str, + expressions: list[str], + owner_type: str = "orgs", + ) -> dict: + """Remove references from a collection.""" + self._require_auth() + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/collections/{collection}/references/" + # DELETE with body + response = self.client.request( + "DELETE", endpoint, json={"references": expressions} + ) + self._handle_error(response) + if response.status_code == 204: + return {} + return response.json() + + def create_expansion( + self, + owner: str, + collection: str, + version: str, + owner_type: str = "orgs", + **params: Any, + ) -> dict: + """Create/trigger an expansion for a collection version.""" + self._require_auth() + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/collections/{collection}/{version}/expansions/" + body = {k: v for k, v in params.items() if v is not None} + return self.post(endpoint, json=body or None) + + def clone( + self, + owner: str, + dest_source: str, + expressions: list[str], + owner_type: str = "orgs", + map_types: Optional[list[str]] = None, + exclude_map_types: Optional[list[str]] = None, + method: Optional[str] = None, + cascade_levels: Optional[str] = None, + cascade_hierarchy: bool = True, + cascade_mappings: bool = True, + ) -> dict: + """Clone concepts using $clone operation.""" + self._require_auth() + _validate_owner_type(owner_type) + endpoint = f"/{owner_type}/{owner}/sources/{dest_source}/$clone/" + + body: dict[str, Any] = {"expressions": expressions} + params: dict[str, Any] = {} + if map_types: + params["mapTypes"] = ",".join(map_types) + if exclude_map_types: + params["excludeMapTypes"] = ",".join(exclude_map_types) + if method: + params["method"] = method + if cascade_levels: + params["cascadeLevels"] = cascade_levels + if not cascade_hierarchy: + params["cascadeHierarchy"] = "false" + if not cascade_mappings: + params["cascadeMappings"] = "false" + + return self.post(endpoint, json=body, params=params or None) + + def set_repo_extra( + self, + owner: str, + repo: str, + key: str, + value: Any, + owner_type: str = "orgs", + repo_type: str = "source", + ) -> dict: + """Set a custom attribute on a repository.""" + self._require_auth() + endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo, suffix=f"extras/{key}/") + return self.put(endpoint, json={key: value}) + + def delete_repo_extra( + self, + owner: str, + repo: str, + key: str, + owner_type: str = "orgs", + repo_type: str = "source", + ) -> dict: + """Delete a custom attribute from a repository.""" + self._require_auth() + _validate_owner_type(owner_type) + stem = _repo_type_stem(repo_type) + return self.delete(f"/{owner_type}/{owner}/{stem}/{repo}/extras/{key}/") + + def get_repo_extras( + self, + owner: str, + repo: str, + owner_type: str = "orgs", + repo_type: str = "source", + ) -> dict: + """Get custom attributes for a repository.""" + endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo, suffix="extras/") + return self.get(endpoint) + + def add_org_member(self, org: str, username: str) -> dict: + """Add a member to an organization.""" + self._require_auth() + return self.put(f"/orgs/{org}/members/{username}/") + + def remove_org_member(self, org: str, username: str) -> dict: + """Remove a member from an organization.""" + self._require_auth() + return self.delete(f"/orgs/{org}/members/{username}/") + + # ── Task operations ───────────────────────────────────────────── + + def list_tasks( + self, + state: Optional[str] = None, + limit: int = 20, + page: int = 1, + ) -> dict: + """List async tasks for the authenticated user.""" + self._require_auth() + params: dict[str, Any] = {"limit": limit, "page": page} + if state: + params["state"] = state + return self._get_list("/tasks/", params=params) + + def get_task(self, task_id: str) -> dict: + """Get details of a specific task.""" + self._require_auth() + return self.get(f"/tasks/{task_id}/") + + # ── Organization CRUD ─────────────────────────────────────────── + + def create_org( + self, + org_id: str, + name: str, + company: Optional[str] = None, + website: Optional[str] = None, + location: Optional[str] = None, + public_access: str = "View", + extras: Optional[dict] = None, + ) -> dict: + """Create a new organization.""" + self._require_auth() + body: dict[str, Any] = { + "id": org_id, + "name": name, + "public_access": public_access, + } + if company: + body["company"] = company + if website: + body["website"] = website + if location: + body["location"] = location + if extras: + body["extras"] = extras + return self.post("/orgs/", json=body) + + def delete_org(self, org: str) -> dict: + """Delete an organization.""" + self._require_auth() + return self.delete(f"/orgs/{org}/") + + # ── Repository delete ─────────────────────────────────────────── + + def delete_repo( + self, + owner: str, + repo: str, + owner_type: str = "orgs", + repo_type: str = "source", + ) -> dict: + """Delete a repository (source or collection).""" + self._require_auth() + endpoint = _build_repo_endpoint(owner_type, owner, repo_type, repo) + return self.delete(endpoint) + + # ── Bulk import ───────────────────────────────────────────────── + + def bulk_import(self, lines: list[dict], update_if_exists: bool = True) -> dict: + """Submit a bulk import (JSON Lines format to /manage/bulkimport/).""" + self._require_auth() + if not lines: + return {"message": "no references"} + import json as _json + body = "\n".join(_json.dumps(item, ensure_ascii=False) for item in lines) + "\n" + params = {} + if update_if_exists: + params["update_if_exists"] = "true" + self._log_request("POST", "/manage/bulkimport/", params) + response = self.client.post( + "/manage/bulkimport/", + content=body.encode("utf-8"), + headers={"Content-Type": "application/json"}, + params=params, + timeout=120.0, + ) + self._handle_error(response) + try: + return response.json() + except Exception: + return {"message": response.text} + + def get_bulk_import_status(self) -> list: + """Get status of bulk import tasks.""" + self._require_auth() + return self.get("/manage/bulkimport/") + + # ── Import (new /importers/bulk-import/ endpoint) ───────────── + + IMPORT_MIME_MAP = { + ".json": "application/json", + ".jsonl": "application/json", + ".csv": "text/csv", + ".zip": "application/zip", + } + + def import_file( + self, + file_path: str, + queue: str | None = None, + update_if_exists: bool = True, + parallel: int | None = None, + ) -> dict: + """Upload a file for bulk import via /importers/bulk-import/. + + Supports .json, .jsonl, .csv, and .zip (OCL export) files. + The API handles format detection and conversion server-side. + """ + import os + + self._require_auth() + + endpoint = "/importers/bulk-import/" + if queue: + endpoint = f"/importers/bulk-import/{queue}/" + + params = {"update_if_exists": "true" if update_if_exists else "false"} + + filename = os.path.basename(file_path) + ext = os.path.splitext(filename)[1].lower() + mime = self.IMPORT_MIME_MAP.get(ext, "application/octet-stream") + + data = {} + if parallel is not None: + data["parallel"] = str(parallel) + + self._log_request("POST", endpoint, params) + with open(file_path, "rb") as f: + response = self.client.post( + endpoint, + files={"file": (filename, f, mime)}, + data=data or None, + params=params, + timeout=120.0, + ) + self._handle_error(response) + try: + return response.json() + except Exception: + return {"message": response.text} + + def import_list(self, queue: str | None = None) -> Any: + """List active/recent bulk imports.""" + self._require_auth() + endpoint = "/importers/bulk-import/" + if queue: + endpoint = f"/importers/bulk-import/{queue}/" + return self.get(endpoint) + + def import_status(self, task_id: str) -> Any: + """Get status of a specific bulk import task.""" + self._require_auth() + return self.get("/importers/bulk-import/", params={"task": task_id}) + + # ── Cascade concept fetching (for pruning) ────────────────────── + + def fetch_cascade_children(self, concept_url: str) -> set[str]: + """Fetch child concept URLs via cascade for a given concept expression.""" + url = concept_url.rstrip("/") + "/cascade/" + params = { + "method": "sourcetoconcepts", + "cascadeHierarchy": "true", + "cascadeMappings": "true", + "cascadeLevels": "*", + "reverse": "false", + "includeRetired": "false", + } + try: + data = self.get(url, params=params) + except (APIError, Exception): + return set() + entries = data.get("entry") or [] + if isinstance(entries, dict): + entries = [entries] + children: set[str] = set() + parent = concept_url.rstrip("/") + "/" + for entry in entries: + if not isinstance(entry, dict) or entry.get("type") != "Concept": + continue + child_url = entry.get("url", "").rstrip("/") + "/" + if child_url and child_url != parent: + children.add(child_url) + return children + + # ── Exports ────────────────────────────────────────────────────── + + def _export_endpoint( + self, owner: str, repo: str, version: str, + owner_type: str, repo_type: str, + ) -> str: + return _build_repo_endpoint( + owner_type, owner, repo_type, repo, version, "export/" + ) + + def export_status( + self, + owner: str, + repo: str, + version: str, + owner_type: str = "orgs", + repo_type: str = "source", + ) -> dict: + """Check export status for a repository version via HEAD request.""" + self._require_auth() + endpoint = self._export_endpoint(owner, repo, version, owner_type, repo_type) + response = self.request("HEAD", endpoint, timeout=45.0) + code = response.status_code + if code in (200, 302): + filename = "" + cd = response.headers.get("content-disposition", "") + if "filename=" in cd: + filename = cd.split("filename=", 1)[1].strip().strip('"') + url = response.headers.get("location", "") + return {"status": "ready", "status_code": code, "filename": filename, "url": url} + elif code == 204: + return {"status": "not_found", "status_code": 204} + elif code == 208: + return {"status": "processing", "status_code": 208} + elif code == 404: + raise APIError("Repository version not found", status_code=404) + else: + self._handle_error(response) + return {"status": "unknown", "status_code": code} + + def export_create( + self, + owner: str, + repo: str, + version: str, + owner_type: str = "orgs", + repo_type: str = "source", + ) -> dict: + """Trigger export creation for a repository version.""" + self._require_auth() + endpoint = self._export_endpoint(owner, repo, version, owner_type, repo_type) + response = self.request("POST", endpoint, timeout=45.0) + code = response.status_code + if code == 202: + return {"status": "accepted", "status_code": 202, + "message": "Export creation started"} + elif code in (200, 303): + return {"status": "already_exists", "status_code": code, + "message": "Export already exists"} + elif code == 409: + return {"status": "conflict", "status_code": 409, + "message": "Export is already being processed"} + else: + self._handle_error(response) + return {"status": "unknown", "status_code": code} + + def export_delete( + self, + owner: str, + repo: str, + version: str, + owner_type: str = "orgs", + repo_type: str = "source", + ) -> dict: + """Delete an export for a repository version.""" + self._require_auth() + endpoint = self._export_endpoint(owner, repo, version, owner_type, repo_type) + response = self.request("DELETE", endpoint, timeout=30.0) + if response.status_code == 204: + return {"status": "deleted", "status_code": 204} + elif response.status_code == 404: + raise APIError("Export not found", status_code=404) + else: + self._handle_error(response) + return {"status": "unknown", "status_code": response.status_code} + + def export_download( + self, + owner: str, + repo: str, + version: str, + owner_type: str = "orgs", + repo_type: str = "source", + ) -> httpx.Response: + """Download an export file. Returns the streaming response. + + The response has Content-Type: application/zip and a + Content-Disposition header with the filename. + Caller should consume and close the response. + """ + self._require_auth() + endpoint = self._export_endpoint(owner, repo, version, owner_type, repo_type) + response = self.request("GET", endpoint, timeout=300.0) + code = response.status_code + if code == 200: + return response + elif code == 302: + # API returns 302 redirect to S3 signed URL — follow it + redirect_url = response.headers.get("location", "") + if not redirect_url: + raise APIError("Export redirect missing Location header", status_code=302) + redirect_response = httpx.get(redirect_url, timeout=300.0) + redirect_response.raise_for_status() + return redirect_response + elif code == 204: + raise APIError( + "Export does not exist. Use 'ocl repo export create' first.", + status_code=204, + ) + elif code == 208: + raise APIError( + "Export is still being processed. Try again shortly.", + status_code=208, + ) + elif code == 404: + raise APIError("Repository version not found", status_code=404) + else: + self._handle_error(response) + raise APIError(f"Unexpected status {code}", status_code=code) diff --git a/tests/test_core_package.py b/tests/test_core_package.py new file mode 100644 index 0000000..cdf366b --- /dev/null +++ b/tests/test_core_package.py @@ -0,0 +1,37 @@ +import builtins +import importlib +import sys +from importlib.metadata import version + +import openconceptlab +from openconceptlab import APIError, OCLAPIClient + + +def test_core_exports_client_error_and_version(): + assert openconceptlab.__version__ == version("openconceptlab") + assert OCLAPIClient.__module__ == "openconceptlab.client" + assert issubclass(APIError, Exception) + + +def test_cli_shim_and_version_point_at_the_core(): + from ocl_cli import CLIENT_NAME, __version__ + from ocl_cli.api_client import APIError as ShimError + from ocl_cli.api_client import OCLAPIClient as ShimClient + + assert ShimClient is OCLAPIClient and ShimError is APIError + assert __version__ == openconceptlab.__version__ + assert CLIENT_NAME == f"ocl-cli/{openconceptlab.__version__}" + + +def test_core_imports_without_click(monkeypatch): + real_import = builtins.__import__ + + def guarded(name, *args, **kwargs): + if name == "click" or name.startswith("click."): + raise ImportError("the core must not need click") + return real_import(name, *args, **kwargs) + + for mod in [m for m in sys.modules if m == "openconceptlab" or m.startswith("openconceptlab.")]: + monkeypatch.delitem(sys.modules, mod) + monkeypatch.setattr(builtins, "__import__", guarded) + importlib.import_module("openconceptlab") From 0592bbb5eb87a13a1daf8ad72825fa62874347c0 Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 09:13:51 -0400 Subject: [PATCH 03/10] OpenConceptLab/ocl_online#507 | Client options: client name, auth scheme, headers, timeout Co-Authored-By: Claude Haiku 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- src/ocl_cli/commands/auth.py | 3 +- src/ocl_cli/main.py | 4 +-- src/openconceptlab/client.py | 32 ++++++++++++++----- tests/test_client.py | 61 ++++++++++++++++++++++++++++++++++++ 4 files changed, 90 insertions(+), 10 deletions(-) create mode 100644 tests/test_client.py diff --git a/src/ocl_cli/commands/auth.py b/src/ocl_cli/commands/auth.py index 3a87834..d0f27fe 100644 --- a/src/ocl_cli/commands/auth.py +++ b/src/ocl_cli/commands/auth.py @@ -26,8 +26,9 @@ def login(ctx): token = token.strip() # Validate the token by calling GET /user/ + from ocl_cli import CLIENT_NAME from ocl_cli.api_client import OCLAPIClient - test_client = OCLAPIClient(base_url=server.base_url, token=token) + test_client = OCLAPIClient(base_url=server.base_url, token=token, client_name=CLIENT_NAME) try: user = test_client.get_user() config.set_token(server.server_id, token) diff --git a/src/ocl_cli/main.py b/src/ocl_cli/main.py index 697c679..589862a 100644 --- a/src/ocl_cli/main.py +++ b/src/ocl_cli/main.py @@ -7,7 +7,7 @@ import click -from ocl_cli import __version__ +from ocl_cli import CLIENT_NAME, __version__ from ocl_cli.api_client import APIError, OCLAPIClient from ocl_cli.config import CLIConfig from ocl_cli.output import output_error @@ -29,7 +29,7 @@ def cli(ctx, json_output, server_id, token, debug, show_request): server = config.get_server(server_id) resolved_token = config.resolve_token(server, token_override=token) - client = OCLAPIClient(base_url=server.base_url, token=resolved_token) + client = OCLAPIClient(base_url=server.base_url, token=resolved_token, client_name=CLIENT_NAME) client.debug = debug client.show_request = show_request diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index 809b331..04b8427 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -70,25 +70,43 @@ class OCLAPIClient: DEFAULT_BASE_URL = "https://api.openconceptlab.org" - def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None): + def __init__( + self, + base_url: Optional[str] = None, + token: Optional[str] = None, + *, + auth_scheme: str = "Token", + client_name: Optional[str] = None, + headers: Optional[dict[str, str]] = None, + timeout: float = 30.0, + ): + """Create a client. + + ``auth_scheme`` is ``"Token"`` for an OCL API token or ``"Bearer"`` for an + OIDC access token. ``client_name`` identifies the calling tool in + ``X-OCL-CLIENT`` and the User-Agent. ``headers`` are sent on every request + and override the defaults. + """ self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/") self.token = token self.debug = False self.show_request = False self._show_request_server_printed = False - headers = { + name = client_name or f"openconceptlab/{__version__}" + default_headers = { "Accept": "application/json", - "X-OCL-CLIENT": f"ocl-cli/{__version__}", - "User-Agent": f"ocl-cli/{__version__} (httpx/{httpx.__version__}; python/{platform.python_version()})", + "X-OCL-CLIENT": name, + "User-Agent": f"{name} (httpx/{httpx.__version__}; python/{platform.python_version()})", } if self.token: - headers["Authorization"] = f"Token {self.token}" + default_headers["Authorization"] = f"{auth_scheme} {self.token}" + default_headers.update(headers or {}) self.client = httpx.Client( base_url=self.base_url, - headers=headers, - timeout=30.0, + headers=default_headers, + timeout=timeout, limits=httpx.Limits(max_keepalive_connections=10, max_connections=20), ) diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..b7cd3d9 --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,61 @@ +from types import SimpleNamespace +from unittest.mock import patch + +from click.testing import CliRunner + +from openconceptlab import OCLAPIClient, __version__ + +BASE = "https://api.example.test" + + +def test_default_identity_and_token_scheme(httpx_mock): + httpx_mock.add_response(json={}) + OCLAPIClient(base_url=BASE, token="abc").get("/x/") + req = httpx_mock.get_request() + assert req.headers["Authorization"] == "Token abc" + assert req.headers["X-OCL-CLIENT"] == f"openconceptlab/{__version__}" + assert req.headers["User-Agent"].startswith(f"openconceptlab/{__version__} (httpx/") + + +def test_bearer_client_name_extra_headers_and_timeout(httpx_mock): + httpx_mock.add_response(json={}) + client = OCLAPIClient( + base_url=BASE, + token="kc-token", + auth_scheme="Bearer", + client_name="ocl-example/1.2.3", + headers={"X-OCL-Request-Source": "api"}, + timeout=12.5, + ) + client.get("/x/") + req = httpx_mock.get_request() + assert req.headers["Authorization"] == "Bearer kc-token" + assert req.headers["X-OCL-CLIENT"] == "ocl-example/1.2.3" + assert req.headers["X-OCL-Request-Source"] == "api" + assert client.client.timeout.read == 12.5 + + +def test_no_authorization_header_without_a_token(httpx_mock): + httpx_mock.add_response(json={}) + OCLAPIClient(base_url=BASE).get("/x/") + assert "Authorization" not in httpx_mock.get_request().headers + + +class _FakeConfig: + def get_server(self, server_id): + return SimpleNamespace(base_url=BASE) + + def resolve_token(self, server, token_override=None): + return "t" + + +def test_cli_names_itself_to_the_api(): + from ocl_cli import CLIENT_NAME + from ocl_cli.main import cli + + with ( + patch("ocl_cli.main.CLIConfig.load", return_value=_FakeConfig()), + patch("ocl_cli.main.OCLAPIClient") as client_cls, + ): + CliRunner().invoke(cli, ["repo", "--help"]) + assert client_cls.call_args.kwargs["client_name"] == CLIENT_NAME From d3ac4a89e20f4627bd6beafc99741ed1ec38373d Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 09:16:08 -0400 Subject: [PATCH 04/10] OpenConceptLab/ocl_online#507 | Retry a POST only when the connection failed; raise the httpx error after retries Co-Authored-By: Claude Haiku 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- src/openconceptlab/client.py | 11 ++++++++++- tests/test_client.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 1 deletion(-) diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index 04b8427..69b661a 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -17,6 +17,9 @@ from openconceptlab import __version__ +# A POST is not idempotent: retry it only when the request never reached the server. +_RETRY_IF_NOT_SENT = retry_if_exception_type((httpx.ConnectError, httpx.ConnectTimeout)) + class APIError(Exception): """Structured API error with status code and detail.""" @@ -163,6 +166,7 @@ def _handle_error(self, response: httpx.Response) -> None: stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=60), retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), + reraise=True, ) def get(self, endpoint: str, params: Optional[dict] = None, timeout: Optional[float] = None) -> Any: self._log_request("GET", endpoint, params) @@ -176,7 +180,8 @@ def get(self, endpoint: str, params: Optional[dict] = None, timeout: Optional[fl @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=60), - retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), + retry=_RETRY_IF_NOT_SENT, + reraise=True, ) def post(self, endpoint: str, json: Optional[dict] = None, params: Optional[dict] = None) -> Any: self._log_request("POST", endpoint, params, body=json) @@ -188,6 +193,7 @@ def post(self, endpoint: str, json: Optional[dict] = None, params: Optional[dict stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=60), retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), + reraise=True, ) def put(self, endpoint: str, json: Optional[dict] = None, params: Optional[dict] = None) -> Any: self._log_request("PUT", endpoint, params, body=json) @@ -199,6 +205,7 @@ def put(self, endpoint: str, json: Optional[dict] = None, params: Optional[dict] stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=60), retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), + reraise=True, ) def patch(self, endpoint: str, json: Optional[dict] = None, params: Optional[dict] = None) -> Any: self._log_request("PATCH", endpoint, params, body=json) @@ -210,6 +217,7 @@ def patch(self, endpoint: str, json: Optional[dict] = None, params: Optional[dic stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=60), retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), + reraise=True, ) def delete(self, endpoint: str, params: Optional[dict] = None) -> Any: self._log_request("DELETE", endpoint, params) @@ -261,6 +269,7 @@ def _normalize(self, response: Any, headers: Optional[dict] = None) -> dict: stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=60), retry=retry_if_exception_type((httpx.TimeoutException, httpx.NetworkError)), + reraise=True, ) def _get_list(self, endpoint: str, params: Optional[dict] = None) -> dict: """GET a list endpoint, preserving pagination headers. diff --git a/tests/test_client.py b/tests/test_client.py index b7cd3d9..287b52d 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,7 +1,10 @@ from types import SimpleNamespace from unittest.mock import patch +import httpx +import pytest from click.testing import CliRunner +from tenacity import wait_none from openconceptlab import OCLAPIClient, __version__ @@ -41,6 +44,31 @@ def test_no_authorization_header_without_a_token(httpx_mock): assert "Authorization" not in httpx_mock.get_request().headers +def test_post_is_not_retried_after_a_read_timeout(httpx_mock, monkeypatch): + monkeypatch.setattr(OCLAPIClient.post.retry, "wait", wait_none()) + httpx_mock.add_exception(httpx.ReadTimeout("slow")) + with pytest.raises(httpx.ReadTimeout): + OCLAPIClient(base_url=BASE, token="t").post("/concepts/$match/", json={"rows": []}) + assert len(httpx_mock.get_requests()) == 1 + + +def test_post_is_retried_when_the_connection_failed(httpx_mock, monkeypatch): + monkeypatch.setattr(OCLAPIClient.post.retry, "wait", wait_none()) + httpx_mock.add_exception(httpx.ConnectError("refused")) + httpx_mock.add_response(method="POST", json={"ok": True}) + assert OCLAPIClient(base_url=BASE, token="t").post("/x/", json={}) == {"ok": True} + assert len(httpx_mock.get_requests()) == 2 + + +def test_get_raises_the_network_error_once_retries_run_out(httpx_mock, monkeypatch): + monkeypatch.setattr(OCLAPIClient.get.retry, "wait", wait_none()) + for _ in range(3): + httpx_mock.add_exception(httpx.ReadTimeout("slow")) + with pytest.raises(httpx.ReadTimeout): + OCLAPIClient(base_url=BASE, token="t").get("/x/") + assert len(httpx_mock.get_requests()) == 3 + + class _FakeConfig: def get_server(self, server_id): return SimpleNamespace(base_url=BASE) From a08663538ead9692ec46e9a2b28e25355db07e7b Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 09:17:36 -0400 Subject: [PATCH 05/10] OpenConceptLab/ocl_online#507 | match_concepts: full rows and the reranker Co-Authored-By: Claude Haiku 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- src/openconceptlab/client.py | 14 ++++++++++++-- tests/test_client.py | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 2 deletions(-) diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index 69b661a..f35ea64 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -767,9 +767,17 @@ def match_concepts( include_mappings: bool = False, semantic: bool = True, verbose: bool = False, + reranker: bool = False, + rows: Optional[list[dict]] = None, ) -> dict: - """Match concepts using the $match endpoint.""" - rows = [{"name": term} for term in terms] + """Match concepts using the $match endpoint. + + ``rows`` sends full rows (a ``name`` plus properties such as + ``concept_class``, ``datatype``, ``description`` or ``synonyms``) instead + of building ``{"name": term}`` rows from ``terms``. ``reranker`` asks OCL + to rerank the candidates. + """ + rows = rows if rows is not None else [{"name": term} for term in terms] body: dict[str, Any] = { "rows": rows, "target_repo_url": target_repo_url, @@ -794,6 +802,8 @@ def match_concepts( params["includeMappings"] = True if verbose: params["verbose"] = True + if reranker: + params["reranker"] = True result = self.post("/concepts/$match/", params=params, json=body) return self._normalize(result) diff --git a/tests/test_client.py b/tests/test_client.py index 287b52d..0eafd9d 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,3 +1,5 @@ +import json +import re from types import SimpleNamespace from unittest.mock import patch @@ -87,3 +89,34 @@ def test_cli_names_itself_to_the_api(): ): CliRunner().invoke(cli, ["repo", "--help"]) assert client_cls.call_args.kwargs["client_name"] == CLIENT_NAME + + +MATCH_URL = re.compile(r"https://api\.example\.test/concepts/\$match/.*") + + +def test_match_sends_full_rows_and_the_reranker(httpx_mock): + httpx_mock.add_response(method="POST", url=MATCH_URL, json=[{"row": {"name": "fever"}, "results": []}]) + out = OCLAPIClient(base_url=BASE, token="t").match_concepts( + [], + "/orgs/CIEL/sources/CIEL/", + rows=[{"name": "fever", "concept_class": "Diagnosis"}], + reranker=True, + limit=3, + ) + req = httpx_mock.get_request() + assert json.loads(req.content) == { + "rows": [{"name": "fever", "concept_class": "Diagnosis"}], + "target_repo_url": "/orgs/CIEL/sources/CIEL/", + } + assert req.url.params["reranker"] == "true" + assert req.url.params["semantic"] == "true" + assert req.url.params["limit"] == "3" + assert out == {"count": 1, "results": [{"row": {"name": "fever"}, "results": []}]} + + +def test_match_terms_still_become_name_rows_without_the_reranker(httpx_mock): + httpx_mock.add_response(method="POST", url=MATCH_URL, json=[]) + OCLAPIClient(base_url=BASE, token="t").match_concepts(["fever", "cough"], "/orgs/CIEL/sources/CIEL/") + req = httpx_mock.get_request() + assert json.loads(req.content)["rows"] == [{"name": "fever"}, {"name": "cough"}] + assert "reranker" not in req.url.params From 4585434a1d7efa0a801fa3d4254550e8fc4a7c2d Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 09:17:50 -0400 Subject: [PATCH 06/10] OpenConceptLab/ocl_online#507 | list_expansion_concepts Co-Authored-By: Claude Haiku 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- src/openconceptlab/client.py | 30 ++++++++++++++++++++++++++++++ tests/test_client.py | 33 ++++++++++++++++++++++++++++++++- 2 files changed, 62 insertions(+), 1 deletion(-) diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index f35ea64..fcdf30a 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -853,6 +853,36 @@ def get_expansion( "expansion_url": None, } + def list_expansion_concepts( + self, + owner: str, + collection: str, + collection_version: str, + expansion_id: Optional[str] = None, + owner_type: str = "orgs", + query: Optional[str] = None, + limit: int = 20, + page: int = 1, + ) -> dict: + """List the concepts in a collection version's expansion. + + Uses the version's default expansion when ``expansion_id`` is not given. + """ + _validate_owner_type(owner_type) + version_endpoint = f"/{owner_type}/{owner}/collections/{collection}/{collection_version}/" + if expansion_id: + expansion_url = f"{version_endpoint}expansions/{expansion_id}/" + else: + expansion_url = self.get(version_endpoint).get("expansion_url") + if not expansion_url: + raise APIError( + "This collection version has no default expansion", status_code=404 + ) + params: dict[str, Any] = {"limit": limit, "page": page} + if query: + params["q"] = query + return self._get_list(f"{expansion_url.rstrip('/')}/concepts/", params=params) + def list_collection_refs( self, owner: str, diff --git a/tests/test_client.py b/tests/test_client.py index 0eafd9d..efd7fda 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -8,7 +8,7 @@ from click.testing import CliRunner from tenacity import wait_none -from openconceptlab import OCLAPIClient, __version__ +from openconceptlab import APIError, OCLAPIClient, __version__ BASE = "https://api.example.test" @@ -120,3 +120,34 @@ def test_match_terms_still_become_name_rows_without_the_reranker(httpx_mock): req = httpx_mock.get_request() assert json.loads(req.content)["rows"] == [{"name": "fever"}, {"name": "cough"}] assert "reranker" not in req.url.params + + +def test_expansion_concepts_by_id(httpx_mock): + httpx_mock.add_response( + url=re.compile(r".*/orgs/PIH/collections/HIV/v1/expansions/e1/concepts/.*"), + json=[{"id": "1"}], + headers={"num_found": "40"}, + ) + out = OCLAPIClient(base_url=BASE).list_expansion_concepts("PIH", "HIV", "v1", expansion_id="e1", query="cd4", limit=5) + req = httpx_mock.get_request() + assert req.url.params["q"] == "cd4" and req.url.params["limit"] == "5" + assert out == {"count": 40, "results": [{"id": "1"}]} + + +def test_expansion_concepts_from_the_default_expansion(httpx_mock): + httpx_mock.add_response( + url=f"{BASE}/orgs/PIH/collections/HIV/v1/", + json={"expansion_url": "/orgs/PIH/collections/HIV/v1/expansions/autoexpand-v1/"}, + ) + httpx_mock.add_response( + url=re.compile(r".*/expansions/autoexpand-v1/concepts/.*"), json=[], headers={"num_found": "0"} + ) + out = OCLAPIClient(base_url=BASE).list_expansion_concepts("PIH", "HIV", "v1") + assert out == {"count": 0, "results": []} + + +def test_expansion_concepts_without_a_default_expansion(httpx_mock): + httpx_mock.add_response(url=f"{BASE}/orgs/PIH/collections/HIV/v1/", json={"expansion_url": None}) + with pytest.raises(APIError) as err: + OCLAPIClient(base_url=BASE).list_expansion_concepts("PIH", "HIV", "v1") + assert err.value.status_code == 404 From e8f43feb970d89728e36230d23fe8d7ce15bb8e6 Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 09:18:02 -0400 Subject: [PATCH 07/10] OpenConceptLab/ocl_online#507 | changelog(): $changelog between two versions Co-Authored-By: Claude Haiku 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- src/openconceptlab/client.py | 27 +++++++++++++++++++++++++++ tests/test_client.py | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+) diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index fcdf30a..1e01656 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -883,6 +883,33 @@ def list_expansion_concepts( params["q"] = query return self._get_list(f"{expansion_url.rstrip('/')}/concepts/", params=params) + def changelog( + self, + repo_type: str, + version1_url: str, + version2_url: str, + verbosity: int = 4, + ) -> dict: + """What changed between two versions of a source or collection ($changelog). + + Returns ``{"status": "pending", ...}`` while OCL is still computing it: + HTTP 202 after OCL's 40-second wait, or 409 when it's already queued. + """ + endpoint = f"/{_repo_type_stem(repo_type)}/$changelog/" + response = self.request( + "POST", + endpoint, + params={"verbosity": verbosity}, + json={"version1": version1_url, "version2": version2_url}, + timeout=60.0, + ) + if response.status_code == 202: + return {"status": "pending", "task": response.json()} + if response.status_code == 409: + return {"status": "pending", "detail": "Already queued"} + self._handle_error(response) + return response.json() + def list_collection_refs( self, owner: str, diff --git a/tests/test_client.py b/tests/test_client.py index efd7fda..1f89e1e 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -151,3 +151,35 @@ def test_expansion_concepts_without_a_default_expansion(httpx_mock): with pytest.raises(APIError) as err: OCLAPIClient(base_url=BASE).list_expansion_concepts("PIH", "HIV", "v1") assert err.value.status_code == 404 + + +CHANGELOG_URL = re.compile(r"https://api\.example\.test/sources/\$changelog/.*") +V1, V2 = "/orgs/CIEL/sources/CIEL/v2024/", "/orgs/CIEL/sources/CIEL/v2025/" + + +def test_changelog_returns_the_result(httpx_mock): + httpx_mock.add_response(method="POST", url=CHANGELOG_URL, json={"meta": {}, "concepts": {"new": {}}}) + out = OCLAPIClient(base_url=BASE, token="t").changelog("source", V1, V2) + req = httpx_mock.get_request() + assert json.loads(req.content) == {"version1": V1, "version2": V2} + assert req.url.params["verbosity"] == "4" + assert out == {"meta": {}, "concepts": {"new": {}}} + + +def test_changelog_still_computing(httpx_mock): + httpx_mock.add_response(method="POST", url=CHANGELOG_URL, status_code=202, json={"id": "task-1", "state": "PENDING"}) + out = OCLAPIClient(base_url=BASE, token="t").changelog("source", V1, V2) + assert out == {"status": "pending", "task": {"id": "task-1", "state": "PENDING"}} + + +def test_changelog_already_queued(httpx_mock): + httpx_mock.add_response(method="POST", url=CHANGELOG_URL, status_code=409, json={"detail": "Already Queued"}) + out = OCLAPIClient(base_url=BASE, token="t").changelog("source", V1, V2) + assert out == {"status": "pending", "detail": "Already queued"} + + +def test_changelog_errors_raise(httpx_mock): + httpx_mock.add_response(method="POST", url=CHANGELOG_URL, status_code=404, json={"detail": "Not found."}) + with pytest.raises(APIError) as err: + OCLAPIClient(base_url=BASE, token="t").changelog("source", V1, V2) + assert err.value.status_code == 404 From cc4b2fd3a00607003ce4dd00b0aa73ba869bc843 Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 09:19:43 -0400 Subject: [PATCH 08/10] OpenConceptLab/ocl_online#507 | 0.2.0: library docs and version Co-Authored-By: Claude Haiku 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- CLAUDE.md | 4 ++-- README.md | 15 +++++++++++++++ pyproject.toml | 2 +- 3 files changed, 18 insertions(+), 3 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index f65af26..94a9a29 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -13,7 +13,7 @@ pip install -e ".[dev]" ## Architecture -- **`src/ocl_cli/api_client.py`** — Sync HTTP client with all high-level API methods. Business logic lives here (not in Click commands) so it can be reused by a future MCP server wrapper. +- **`src/openconceptlab/client.py`** — The API client is the open core package `openconceptlab` (`src/openconceptlab/client.py`). `ocl_cli.api_client` is a shim. Put API logic in the core; keep Click and output formatting in `ocl_cli`. - **`src/ocl_cli/config.py`** — Server registry and auth token resolution. Config stored at `~/.ocl/config.json`. - **`src/ocl_cli/output.py`** — JSON and human-readable table formatters. - **`src/ocl_cli/main.py`** — Click root group, global options, command registration. @@ -21,7 +21,7 @@ pip install -e ".[dev]" ## Conventions -- Click commands are thin wrappers. All API logic goes in `api_client.py`. +- Click commands are thin wrappers. All API logic goes in the core client, `src/openconceptlab/client.py`. - Default `owner_type` is `"orgs"` for targeted operations, `"all"` for search/list. - Default `repo_type` is `"all"` for search/list, required for targeted operations. - `--json` / `-j` global flag outputs raw API JSON (unmodified) for agent consumption. diff --git a/README.md b/README.md index f39c4ab..0f56e65 100644 --- a/README.md +++ b/README.md @@ -50,6 +50,21 @@ ocl --version ocl --help ``` +## Use as a Python library + +The `openconceptlab` package also installs the API client the `ocl` command uses: + +```python +from openconceptlab import OCLAPIClient + +with OCLAPIClient(token="", client_name="my-tool/1.0") as ocl: + hits = ocl.search_concepts(query="malaria", owner="CIEL", repo="CIEL", limit=5) + for concept in hits["results"]: + print(concept["id"], concept["display_name"]) +``` + +Pass `auth_scheme="Bearer"` to use an OIDC access token instead of an API token. + ## Quick Start ### 1. Authenticate diff --git a/pyproject.toml b/pyproject.toml index d05f893..ed6a330 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "openconceptlab" version = "0.2.0" -description = "Command-line interface for the Open Concept Lab API" +description = "Python client and command-line interface for the Open Concept Lab API" readme = "README.md" license = "MPL-2.0" requires-python = ">=3.10" From 51691856565f43accebf5260910137377b6a68e0 Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 09:30:16 -0400 Subject: [PATCH 09/10] OpenConceptLab/ocl_online#507 | Case-insensitive header overrides, APIError.body, error docs, CI permissions Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- .github/workflows/ci.yml | 3 +++ README.md | 2 ++ src/openconceptlab/client.py | 37 ++++++++++++++++++++++-------------- tests/test_client.py | 29 +++++++++++++++++++++++++++- 4 files changed, 56 insertions(+), 15 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d582821..a5537ef 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -5,6 +5,9 @@ on: push: branches: [main] +permissions: + contents: read + jobs: test: runs-on: ubuntu-latest diff --git a/README.md b/README.md index 0f56e65..07ce642 100644 --- a/README.md +++ b/README.md @@ -65,6 +65,8 @@ with OCLAPIClient(token="", client_name="my-tool/1.0") as oc Pass `auth_scheme="Bearer"` to use an OIDC access token instead of an API token. +HTTP errors raise `openconceptlab.APIError`, with `status_code`, `detail` and the parsed response `body`. Network failures raise the httpx exception (`httpx.TimeoutException` or `httpx.TransportError`) once retries run out. A client holds one token, so create one per user. + ## Quick Start ### 1. Authenticate diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index 1e01656..4a4ff8d 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -22,12 +22,17 @@ class APIError(Exception): - """Structured API error with status code and detail.""" + """Structured API error with status code and detail. - def __init__(self, message: str, status_code: int = 0, detail: str = ""): + ``body`` is the error response's parsed JSON, or its text when it isn't JSON; + it is None when the error didn't come from an HTTP response. + """ + + def __init__(self, message: str, status_code: int = 0, detail: str = "", body: Any = None): super().__init__(message) self.status_code = status_code self.detail = detail + self.body = body def to_dict(self) -> dict: d = {"error": str(self), "status_code": self.status_code} @@ -104,11 +109,12 @@ def __init__( } if self.token: default_headers["Authorization"] = f"{auth_scheme} {self.token}" - default_headers.update(headers or {}) + merged = httpx.Headers(default_headers) + merged.update(headers or {}) self.client = httpx.Client( base_url=self.base_url, - headers=default_headers, + headers=merged, timeout=timeout, limits=httpx.Limits(max_keepalive_connections=10, max_connections=20), ) @@ -145,22 +151,25 @@ def _log_request(self, method: str, endpoint: str, params: dict | None = None, def _handle_error(self, response: httpx.Response) -> None: """Convert HTTP errors to APIError.""" + if response.status_code < 400: + return + try: + body = response.json() + except Exception: + body = response.text if response.status_code == 429: retry_after = response.headers.get("Retry-After", "60") raise APIError( f"Rate limit exceeded. Retry after {retry_after}s", status_code=429, + body=body, ) - if response.status_code >= 400: - try: - detail = response.json() - except Exception: - detail = response.text - raise APIError( - f"HTTP {response.status_code}", - status_code=response.status_code, - detail=str(detail), - ) + raise APIError( + f"HTTP {response.status_code}", + status_code=response.status_code, + detail=str(body), + body=body, + ) @retry( stop=stop_after_attempt(3), diff --git a/tests/test_client.py b/tests/test_client.py index 1f89e1e..cafa28d 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -40,6 +40,32 @@ def test_bearer_client_name_extra_headers_and_timeout(httpx_mock): assert client.client.timeout.read == 12.5 +def test_extra_headers_override_defaults_case_insensitively(httpx_mock): + httpx_mock.add_response(json={}) + OCLAPIClient( + base_url=BASE, token="abc", headers={"authorization": "Bearer other", "x-ocl-client": "mine/1"} + ).get("/x/") + req = httpx_mock.get_request() + assert req.headers.get_list("Authorization") == ["Bearer other"] + assert req.headers.get_list("X-OCL-CLIENT") == ["mine/1"] + + +def test_api_error_carries_the_parsed_body(httpx_mock): + httpx_mock.add_response(status_code=404, json={"detail": "Not found."}) + with pytest.raises(APIError) as err: + OCLAPIClient(base_url=BASE).get("/x/") + assert err.value.body == {"detail": "Not found."} + assert err.value.detail == "{'detail': 'Not found.'}" # unchanged: the CLI prints it + + +def test_api_error_body_is_the_text_when_not_json(httpx_mock): + httpx_mock.add_response(status_code=502, text="Bad gateway") + with pytest.raises(APIError) as err: + OCLAPIClient(base_url=BASE).get("/x/") + assert err.value.body == "Bad gateway" + assert err.value.status_code == 502 + + def test_no_authorization_header_without_a_token(httpx_mock): httpx_mock.add_response(json={}) OCLAPIClient(base_url=BASE).get("/x/") @@ -87,7 +113,8 @@ def test_cli_names_itself_to_the_api(): patch("ocl_cli.main.CLIConfig.load", return_value=_FakeConfig()), patch("ocl_cli.main.OCLAPIClient") as client_cls, ): - CliRunner().invoke(cli, ["repo", "--help"]) + result = CliRunner().invoke(cli, ["repo", "--help"]) + assert result.exit_code == 0, result.output assert client_cls.call_args.kwargs["client_name"] == CLIENT_NAME From a5441264337d0bd170da6b593c939452e7fdb74b Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Wed, 21 Oct 2026 09:48:43 -0400 Subject: [PATCH 10/10] OpenConceptLab/ocl_online#507 | Send requests only to the API's own origin A request hook on the httpx client refuses any request outside base_url's origin and path, including URLs taken from API responses, so the Authorization header only ever goes to the API. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01QvHhCgtEqAD6xKyJAbToDH --- src/openconceptlab/client.py | 15 +++++++++++++++ tests/test_client.py | 21 +++++++++++++++++++++ 2 files changed, 36 insertions(+) diff --git a/src/openconceptlab/client.py b/src/openconceptlab/client.py index 4a4ff8d..b427af7 100644 --- a/src/openconceptlab/client.py +++ b/src/openconceptlab/client.py @@ -96,6 +96,7 @@ def __init__( and override the defaults. """ self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/") + self._base = httpx.URL(self.base_url) self.token = token self.debug = False self.show_request = False @@ -117,8 +118,22 @@ def __init__( headers=merged, timeout=timeout, limits=httpx.Limits(max_keepalive_connections=10, max_connections=20), + event_hooks={"request": [self._check_destination]}, ) + def _check_destination(self, request: httpx.Request) -> None: + """Refuse any request outside the API's origin and path: the token goes only there.""" + base, url = self._base, request.url + prefix = base.path.rstrip("/") + if ( + (url.scheme, url.host, url.port) != (base.scheme, base.host, base.port) + or (prefix and url.path != prefix and not url.path.startswith(prefix + "/")) + ): + raise APIError( + f"Refusing to send a request outside {base.scheme}://{base.host}: {url.host}", + status_code=0, + ) + def close(self): self.client.close() diff --git a/tests/test_client.py b/tests/test_client.py index cafa28d..4e0bf7a 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -210,3 +210,24 @@ def test_changelog_errors_raise(httpx_mock): with pytest.raises(APIError) as err: OCLAPIClient(base_url=BASE, token="t").changelog("source", V1, V2) assert err.value.status_code == 404 + + +def test_expansion_url_on_another_host_is_refused(httpx_mock): + httpx_mock.add_response( + url=f"{BASE}/orgs/PIH/collections/HIV/v1/", + json={"expansion_url": "https://elsewhere.example/orgs/PIH/collections/HIV/v1/expansions/e1/"}, + ) + with pytest.raises(APIError): + OCLAPIClient(base_url=BASE, token="t").list_expansion_concepts("PIH", "HIV", "v1") + assert len(httpx_mock.get_requests()) == 1 + + +def test_relative_path_cannot_leave_base_url_prefix(httpx_mock): + with pytest.raises(APIError): + OCLAPIClient(base_url="https://api.example.test/ocl-api").get("../other/") + assert httpx_mock.get_requests() == [] + + +def test_same_origin_absolute_url_is_allowed(httpx_mock): + httpx_mock.add_response(url=f"{BASE}/x/", json={"ok": True}) + assert OCLAPIClient(base_url=BASE).get(f"{BASE}/x/") == {"ok": True}