open-project/skills/openproject/scripts/opclient.py

#!/usr/bin/env python3
"""
OpenProject API v3 client — standard library only (no pip installs).

Talks to an OpenProject instance over its HAL+JSON REST API.
Auth: API token via HTTP Basic auth (username "apikey", password = token).

Config resolution order (first hit wins for each key):
  1. Environment variables  OPENPROJECT_URL / OPENPROJECT_API_KEY
  2. config.json next to the plugin root (../config.json relative to this file)
  3. ~/.config/openproject/config.json

This module is imported by op.py (the CLI) and can also be used directly:

    from opclient import OpenProjectClient
    op = OpenProjectClient.from_config()
    me = op.get("/api/v3/users/me")
"""

from __future__ import annotations

import base64
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request

__all__ = ["OpenProjectClient", "OpenProjectError", "load_config"]

DEFAULT_TIMEOUT = 30


class OpenProjectError(Exception):
    """Raised for API and configuration errors. Carries status + parsed body."""

    def __init__(self, message, status=None, body=None):
        super().__init__(message)
        self.status = status
        self.body = body


# --------------------------------------------------------------------------- #
# Configuration
# --------------------------------------------------------------------------- #

def _config_search_paths():
    """Look for config.json at the plugin root and common locations.

    Scripts live at <root>/skills/openproject/scripts/, so we walk up a few
    parent directories to find the plugin root regardless of install depth.
    """
    paths = []
    if os.environ.get("CLAUDE_PLUGIN_ROOT"):
        paths.append(os.path.join(os.environ["CLAUDE_PLUGIN_ROOT"], "config.json"))
    cur = os.path.dirname(os.path.abspath(__file__))
    for _ in range(5):  # scripts -> openproject -> skills -> <root> -> ...
        paths.append(os.path.join(cur, "config.json"))
        cur = os.path.dirname(cur)
    paths.append(os.path.expanduser("~/.config/openproject/config.json"))
    seen, ordered = set(), []
    for p in paths:
        if p not in seen:
            seen.add(p)
            ordered.append(p)
    return ordered


def load_config():
    """Merge config sources. Env vars override file values."""
    cfg = {"url": None, "api_key": None, "default_project": None}

    for path in _config_search_paths():
        if os.path.isfile(path):
            try:
                with open(path, "r", encoding="utf-8") as fh:
                    data = json.load(fh)
                for key in cfg:
                    if data.get(key) not in (None, "", "REPLACE_WITH_YOUR_API_TOKEN"):
                        cfg[key] = data[key]
            except (OSError, ValueError) as exc:
                print(f"warning: could not read {path}: {exc}", file=sys.stderr)
            break  # first existing config file wins

    # Environment overrides
    if os.environ.get("OPENPROJECT_URL"):
        cfg["url"] = os.environ["OPENPROJECT_URL"]
    if os.environ.get("OPENPROJECT_API_KEY"):
        cfg["api_key"] = os.environ["OPENPROJECT_API_KEY"]
    if os.environ.get("OPENPROJECT_DEFAULT_PROJECT"):
        cfg["default_project"] = os.environ["OPENPROJECT_DEFAULT_PROJECT"]

    return cfg


# --------------------------------------------------------------------------- #
# Client
# --------------------------------------------------------------------------- #

class OpenProjectClient:
    def __init__(self, url, api_key, default_project=None, timeout=DEFAULT_TIMEOUT):
        if not url:
            raise OpenProjectError(
                "No OpenProject URL configured. Set OPENPROJECT_URL or config.json 'url'."
            )
        if not api_key:
            raise OpenProjectError(
                "No API key configured. Set OPENPROJECT_API_KEY or config.json 'api_key'."
            )
        self.base_url = url.rstrip("/")
        self.api_key = api_key
        self.default_project = default_project
        self.timeout = timeout
        token = base64.b64encode(f"apikey:{api_key}".encode()).decode()
        self._auth_header = f"Basic {token}"

    @classmethod
    def from_config(cls, timeout=DEFAULT_TIMEOUT):
        cfg = load_config()
        return cls(cfg["url"], cfg["api_key"], cfg.get("default_project"), timeout)

    # -- low-level HTTP ----------------------------------------------------- #

    def _request(self, method, path, params=None, body=None):
        # Allow callers to pass either "/api/v3/..." or a full URL.
        if path.startswith("http://") or path.startswith("https://"):
            url = path
        else:
            url = self.base_url + path
        if params:
            # Drop None values; JSON-encode any non-string (filters/sortBy).
            clean = {}
            for k, v in params.items():
                if v is None:
                    continue
                clean[k] = v if isinstance(v, str) else json.dumps(v)
            url += "?" + urllib.parse.urlencode(clean)

        data = None
        headers = {
            "Authorization": self._auth_header,
            "Accept": "application/json",
        }
        if body is not None:
            data = json.dumps(body).encode("utf-8")
            headers["Content-Type"] = "application/json"

        req = urllib.request.Request(url, data=data, headers=headers, method=method)
        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
                raw = resp.read()
                if not raw:
                    return {}
                return json.loads(raw.decode("utf-8"))
        except urllib.error.HTTPError as exc:
            raw = exc.read().decode("utf-8", "replace")
            parsed = None
            try:
                parsed = json.loads(raw)
            except ValueError:
                pass
            msg = self._format_api_error(exc.code, parsed, raw)
            raise OpenProjectError(msg, status=exc.code, body=parsed or raw)
        except urllib.error.URLError as exc:
            raise OpenProjectError(
                f"Could not reach {self.base_url} ({exc.reason}). "
                f"Check the URL and network."
            )

    @staticmethod
    def _format_api_error(code, parsed, raw):
        if isinstance(parsed, dict):
            message = parsed.get("message", "")
            details = []
            embedded = parsed.get("_embedded", {})
            for err in embedded.get("errors", []) or []:
                if isinstance(err, dict) and err.get("message"):
                    details.append(err["message"])
            if details:
                message = message + " | " + "; ".join(details)
            if message:
                return f"HTTP {code}: {message}"
        return f"HTTP {code}: {raw[:400]}"

    def get(self, path, params=None):
        return self._request("GET", path, params=params)

    def post(self, path, body=None, params=None):
        return self._request("POST", path, params=params, body=body or {})

    def patch(self, path, body=None, params=None):
        return self._request("PATCH", path, params=params, body=body or {})

    def delete(self, path, params=None):
        return self._request("DELETE", path, params=params)

    # -- pagination --------------------------------------------------------- #

    def get_all(self, path, params=None, max_items=1000):
        """Follow nextByOffset links, accumulating collection elements."""
        params = dict(params or {})
        params.setdefault("pageSize", 100)
        params.setdefault("offset", 1)
        items = []
        while True:
            page = self.get(path, params=params)
            embedded = page.get("_embedded", {})
            elements = embedded.get("elements", [])
            items.extend(elements)
            if len(items) >= max_items:
                return items[:max_items]
            total = page.get("total", 0)
            if len(items) >= total or not elements:
                return items
            params["offset"] = int(params["offset"]) + 1

    # -- helpers / resolvers ------------------------------------------------ #

    @staticmethod
    def id_from_href(href):
        if not href:
            return None
        return href.rstrip("/").split("/")[-1]

    @staticmethod
    def link_title(resource, link_name):
        link = (resource.get("_links") or {}).get(link_name) or {}
        return link.get("title")

    def me(self):
        return self.get("/api/v3/users/me")

    def find_project(self, ref):
        """Resolve a project by numeric id, identifier, or (partial) name."""
        if ref is None:
            return None
        ref = str(ref)
        if ref.isdigit():
            return self.get(f"/api/v3/projects/{ref}")
        # Try identifier lookup directly.
        try:
            return self.get(f"/api/v3/projects/{urllib.parse.quote(ref)}")
        except OpenProjectError:
            pass
        # Fall back to name search.
        flt = [{"name_and_identifier": {"operator": "~", "values": [ref]}}]
        results = self.get_all("/api/v3/projects", params={"filters": flt}, max_items=50)
        if not results:
            raise OpenProjectError(f"No project matching '{ref}'.")
        return results[0]

    def find_user(self, ref):
        """Resolve a user by 'me', numeric id, login, or name fragment."""
        if ref is None:
            return None
        ref = str(ref)
        if ref.lower() == "me":
            return self.me()
        if ref.isdigit():
            return self.get(f"/api/v3/users/{ref}")
        flt = [{"name": {"operator": "~", "values": [ref]}}]
        results = self.get_all("/api/v3/users", params={"filters": flt}, max_items=25)
        if not results:
            raise OpenProjectError(f"No user matching '{ref}'.")
        return results[0]

    def _resolve_by_name(self, collection_path, ref, fields=("name",)):
        """Generic resolver for small collections (types, statuses, priorities)."""
        ref = str(ref)
        if ref.isdigit():
            return self.get(f"{collection_path}/{ref}")
        for item in self.get_all(collection_path, max_items=200):
            for field in fields:
                val = item.get(field)
                if val and val.lower() == ref.lower():
                    return item
        # second pass: substring match
        for item in self.get_all(collection_path, max_items=200):
            for field in fields:
                val = item.get(field)
                if val and ref.lower() in val.lower():
                    return item
        raise OpenProjectError(f"No item in {collection_path} matching '{ref}'.")

    def find_type(self, ref):
        return self._resolve_by_name("/api/v3/types", ref)

    def find_status(self, ref):
        return self._resolve_by_name("/api/v3/statuses", ref)

    def find_priority(self, ref):
        return self._resolve_by_name("/api/v3/priorities", ref)

    def find_role(self, ref):
        return self._resolve_by_name("/api/v3/roles", ref)

    # -- create / relate helpers ------------------------------------------- #

    @staticmethod
    def slugify(name):
        """Make a valid OpenProject project identifier from a name."""
        slug = re.sub(r"[^a-z0-9]+", "-", str(name).lower()).strip("-")
        return slug[:100] or "project"

    def create_project(self, name, identifier=None, description=None, parent=None):
        body = {"name": name, "identifier": identifier or self.slugify(name)}
        if description:
            body["description"] = {"format": "markdown", "raw": description}
        if parent:
            pid = self.find_project(parent)["id"]
            body["_links"] = {"parent": {"href": f"/api/v3/projects/{pid}"}}
        return self.post("/api/v3/projects", body=body)

    def add_membership(self, project_id, user_id, role_ref):
        role = self.find_role(role_ref)
        body = {
            "_links": {
                "project": {"href": f"/api/v3/projects/{project_id}"},
                "principal": {"href": f"/api/v3/users/{user_id}"},
                "roles": [{"href": f"/api/v3/roles/{role['id']}"}],
            }
        }
        return self.post("/api/v3/memberships", body=body)

    # OpenProject relation types: relates, duplicates, blocks, blocked, precedes,
    # follows, includes, partof, requires, required.
    def create_relation(self, from_id, to_id, rel_type="relates", description=None):
        body = {
            "type": rel_type,
            "_links": {"to": {"href": f"/api/v3/work_packages/{to_id}"}},
        }
        if description:
            body["description"] = description
        return self.post(f"/api/v3/work_packages/{from_id}/relations", body=body)

    def create_work_package(self, body, notify=False):
        """Create a WP; falls back to create-without-status + PATCH when the
        workflow rejects setting a non-default status on creation."""
        params = {"notify": "false"} if not notify else None
        try:
            return self.post("/api/v3/work_packages", body=body, params=params)
        except OpenProjectError as exc:
            links = body.get("_links", {})
            if exc.status in (422,) and "status" in links:
                status_link = links.pop("status")
                created = self.post("/api/v3/work_packages", body=body, params=params)
                patch = {"lockVersion": created.get("lockVersion"),
                         "_links": {"status": status_link}}
                try:
                    return self.patch(
                        f"/api/v3/work_packages/{created['id']}", body=patch)
                except OpenProjectError:
                    return created  # created at default status; status change rejected
            raise