feat: add .kc metadata and dependency resolution API methods

Add 5 new SiloClient methods for .kc file server integration: - get_metadata(pn): GET /items/{pn}/metadata - update_metadata(pn, fields): PUT /items/{pn}/metadata - patch_lifecycle(pn, state): PATCH /items/{pn}/metadata/lifecycle - patch_tags(pn, tags): PATCH /items/{pn}/metadata/tags - resolve_dependencies(pn): GET /items/{pn}/dependencies/resolve These endpoints are already implemented server-side and needed by the Create module's silo viewer widgets for live data sync. Refs: kindred/silo-mod#43
2026-02-18 19:35:09 -06:00
1 changed files with 115 additions and 27 deletions
--- a/silo_client/init.py
+++ b/silo_client/init.py
@@ -53,7 +53,9 @@ class SiloSettings:
        """Return path to a custom CA certificate file, or ``""``."""
        raise NotImplementedError

-    def save_auth(self, username: str, role: str = "", source: str = "", token: str = ""):
+    def save_auth(
+        self, username: str, role: str = "", source: str = "", token: str = ""
+    ):
        """Persist authentication info after a successful login."""
        raise NotImplementedError

@@ -258,7 +260,9 @@ class SiloClient:
        except urllib.error.URLError as e:
            raise RuntimeError(f"Connection error: {e.reason}")

-    def _upload_ods(self, path: str, ods_bytes: bytes, filename: str = "upload.ods") -> Any:
+    def _upload_ods(
+        self, path: str, ods_bytes: bytes, filename: str = "upload.ods"
+    ) -> Any:
        """POST an ODS file as multipart/form-data."""
        boundary = "----SiloCalcUpload" + str(abs(hash(filename)))[-8:]
        parts: list = []
@@ -289,7 +293,9 @@ class SiloClient:
        except urllib.error.URLError as e:
            raise RuntimeError(f"Connection error: {e.reason}")

-    def _upload_csv(self, path: str, csv_bytes: bytes, filename: str = "import.csv") -> Any:
+    def _upload_csv(
+        self, path: str, csv_bytes: bytes, filename: str = "import.csv"
+    ) -> Any:
        """POST a CSV file as multipart/form-data."""
        boundary = "----SiloCsvUpload" + str(abs(hash(filename)))[-8:]
        parts: list = []
@@ -322,7 +328,9 @@ class SiloClient:

    # -- Authentication -----------------------------------------------------

-    def login(self, username: str, password: str, token_name: str = "") -> Dict[str, Any]:
+    def login(
+        self, username: str, password: str, token_name: str = ""
+    ) -> Dict[str, Any]:
        """Session login and create a persistent API token.

        Performs: ``POST /login`` (session) -> ``GET /api/auth/me`` (verify)
@@ -342,7 +350,9 @@ class SiloClient:

        # Step 1: POST credentials to /login
        login_url = f"{self._origin}/login"
-        form_data = urllib.parse.urlencode({"username": username, "password": password}).encode()
+        form_data = urllib.parse.urlencode(
+            {"username": username, "password": password}
+        ).encode()
        req = urllib.request.Request(
            login_url,
            data=form_data,
@@ -421,7 +431,9 @@ class SiloClient:
        """List API tokens for the current user."""
        return self._request("GET", "/auth/tokens")

-    def create_token(self, name: str, expires_in_days: Optional[int] = None) -> Dict[str, Any]:
+    def create_token(
+        self, name: str, expires_in_days: Optional[int] = None
+    ) -> Dict[str, Any]:
        """Create a new API token."""
        data: Dict[str, Any] = {"name": name}
        if expires_in_days is not None:
@@ -449,7 +461,9 @@ class SiloClient:
        url = f"{self._origin}/health"
        req = urllib.request.Request(url, method="GET")
        try:
-            with urllib.request.urlopen(req, context=self._ssl_context(), timeout=5) as resp:
+            with urllib.request.urlopen(
+                req, context=self._ssl_context(), timeout=5
+            ) as resp:
                return True, f"OK ({resp.status})"
        except urllib.error.HTTPError as e:
            return True, f"Server error ({e.code})"
@@ -461,7 +475,9 @@ class SiloClient:
    # -- Items --------------------------------------------------------------

    def get_item(self, part_number: str) -> Dict[str, Any]:
-        return self._request("GET", f"/items/{urllib.parse.quote(part_number, safe='')}")
+        return self._request(
+            "GET", f"/items/{urllib.parse.quote(part_number, safe='')}"
+        )

    def list_items(
        self,
@@ -517,11 +533,15 @@ class SiloClient:
    def search_items(self, query: str, limit: int = 100, offset: int = 0) -> list:
        """Full-text search across items."""
        q = urllib.parse.quote(query)
-        return self._request("GET", f"/items/search?q={q}&limit={limit}&offset={offset}")
+        return self._request(
+            "GET", f"/items/search?q={q}&limit={limit}&offset={offset}"
+        )

    def get_item_by_uuid(self, uuid: str) -> Dict[str, Any]:
        """Get an item by its internal UUID."""
-        return self._request("GET", f"/items/by-uuid/{urllib.parse.quote(uuid, safe='')}")
+        return self._request(
+            "GET", f"/items/by-uuid/{urllib.parse.quote(uuid, safe='')}"
+        )

    def delete_item(self, part_number: str) -> None:
        """Delete an item by part number."""
@@ -582,7 +602,9 @@ class SiloClient:
            f"/items/{urllib.parse.quote(part_number, safe='')}/revisions/{revision}",
        )

-    def compare_revisions(self, part_number: str, from_rev: int, to_rev: int) -> Dict[str, Any]:
+    def compare_revisions(
+        self, part_number: str, from_rev: int, to_rev: int
+    ) -> Dict[str, Any]:
        pn = urllib.parse.quote(part_number, safe="")
        return self._request(
            "GET",
@@ -650,6 +672,39 @@ class SiloClient:
        except Exception:
            return None

+    # -- .kc Metadata -------------------------------------------------------
+
+    def get_metadata(self, part_number: str) -> Dict[str, Any]:
+        """Get indexed .kc metadata for an item."""
+        pn = urllib.parse.quote(part_number, safe="")
+        return self._request("GET", f"/items/{pn}/metadata")
+
+    def update_metadata(
+        self, part_number: str, fields: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Update .kc metadata fields."""
+        pn = urllib.parse.quote(part_number, safe="")
+        return self._request("PUT", f"/items/{pn}/metadata", fields)
+
+    def patch_lifecycle(self, part_number: str, state: str) -> Dict[str, Any]:
+        """Transition lifecycle state."""
+        pn = urllib.parse.quote(part_number, safe="")
+        return self._request(
+            "PATCH", f"/items/{pn}/metadata/lifecycle", {"state": state}
+        )
+
+    def patch_tags(self, part_number: str, tags: list) -> Dict[str, Any]:
+        """Add/remove tags on .kc metadata."""
+        pn = urllib.parse.quote(part_number, safe="")
+        return self._request("PATCH", f"/items/{pn}/metadata/tags", {"tags": tags})
+
+    # -- .kc Dependencies ---------------------------------------------------
+
+    def resolve_dependencies(self, part_number: str) -> list:
+        """Resolve dependency UUIDs to part numbers and file availability."""
+        pn = urllib.parse.quote(part_number, safe="")
+        return self._request("GET", f"/items/{pn}/dependencies/resolve")
+
    # -- BOM / Relationships ------------------------------------------------

    def get_bom(self, part_number: str) -> list:
@@ -734,12 +789,16 @@ class SiloClient:
        pn = urllib.parse.quote(part_number, safe="")
        return self._download(f"/items/{pn}/bom/export.csv")

-    def import_bom_csv(self, part_number: str, csv_bytes: bytes, filename: str = "bom.csv") -> Any:
+    def import_bom_csv(
+        self, part_number: str, csv_bytes: bytes, filename: str = "bom.csv"
+    ) -> Any:
        """Import a BOM from a CSV file."""
        pn = urllib.parse.quote(part_number, safe="")
        return self._upload_csv(f"/items/{pn}/bom/import", csv_bytes, filename)

-    def merge_bom(self, part_number: str, ods_bytes: bytes, filename: str = "bom.ods") -> Any:
+    def merge_bom(
+        self, part_number: str, ods_bytes: bytes, filename: str = "bom.ods"
+    ) -> Any:
        """Merge a BOM from an ODS spreadsheet."""
        pn = urllib.parse.quote(part_number, safe="")
        return self._upload_ods(f"/items/{pn}/bom/merge", ods_bytes, filename)
@@ -783,7 +842,9 @@ class SiloClient:
            data["label"] = label
        return self._request("POST", f"/schemas/{s}/segments/{seg}/values", data)

-    def update_enum_value(self, schema: str, segment: str, code: str, **fields) -> Dict[str, Any]:
+    def update_enum_value(
+        self, schema: str, segment: str, code: str, **fields
+    ) -> Dict[str, Any]:
        """Update an enum value in a schema segment."""
        s = urllib.parse.quote(schema, safe="")
        seg = urllib.parse.quote(segment, safe="")
@@ -806,7 +867,9 @@ class SiloClient:
        """Get a single project by code."""
        return self._request("GET", f"/projects/{urllib.parse.quote(code, safe='')}")

-    def create_project(self, code: str, name: str, description: str = "") -> Dict[str, Any]:
+    def create_project(
+        self, code: str, name: str, description: str = ""
+    ) -> Dict[str, Any]:
        """Create a new project."""
        data: Dict[str, Any] = {"code": code, "name": name}
        if description:
@@ -814,15 +877,21 @@ class SiloClient:
        return self._request("POST", "/projects", data)

    def get_project_items(self, code: str) -> list:
-        return self._request("GET", f"/projects/{urllib.parse.quote(code, safe='')}/items")
+        return self._request(
+            "GET", f"/projects/{urllib.parse.quote(code, safe='')}/items"
+        )

    def get_item_projects(self, part_number: str) -> list:
        pn = urllib.parse.quote(part_number, safe="")
        return self._request("GET", f"/items/{pn}/projects")

-    def add_item_projects(self, part_number: str, project_codes: List[str]) -> Dict[str, Any]:
+    def add_item_projects(
+        self, part_number: str, project_codes: List[str]
+    ) -> Dict[str, Any]:
        pn = urllib.parse.quote(part_number, safe="")
-        return self._request("POST", f"/items/{pn}/projects", {"projects": project_codes})
+        return self._request(
+            "POST", f"/items/{pn}/projects", {"projects": project_codes}
+        )

    def update_project(self, code: str, **fields) -> Dict[str, Any]:
        """Update a project's fields."""
@@ -834,7 +903,9 @@ class SiloClient:

    def delete_project(self, code: str) -> None:
        """Delete a project."""
-        self._request("DELETE", f"/projects/{urllib.parse.quote(code, safe='')}", raw=True)
+        self._request(
+            "DELETE", f"/projects/{urllib.parse.quote(code, safe='')}", raw=True
+        )

    def remove_item_project(self, part_number: str, project_code: str) -> None:
        """Remove a project tag from an item."""
@@ -861,7 +932,9 @@ class SiloClient:
        code = urllib.parse.quote(project_code, safe="")
        return self._download(f"/projects/{code}/sheet.ods")

-    def upload_sheet_diff(self, ods_bytes: bytes, filename: str = "sheet.ods") -> Dict[str, Any]:
+    def upload_sheet_diff(
+        self, ods_bytes: bytes, filename: str = "sheet.ods"
+    ) -> Dict[str, Any]:
        return self._upload_ods("/sheets/diff", ods_bytes, filename)

    # -- Files / Attachments ------------------------------------------------
@@ -910,9 +983,14 @@ class SiloClient:
        """Set the thumbnail image for an item (binary PUT)."""
        pn = urllib.parse.quote(part_number, safe="")
        url = f"{self.base_url}/items/{pn}/thumbnail"
-        headers = {"Content-Type": content_type, "Content-Length": str(len(image_bytes))}
+        headers = {
+            "Content-Type": content_type,
+            "Content-Length": str(len(image_bytes)),
+        }
        headers.update(self._auth_headers())
-        req = urllib.request.Request(url, data=image_bytes, headers=headers, method="PUT")
+        req = urllib.request.Request(
+            url, data=image_bytes, headers=headers, method="PUT"
+        )
        try:
            with urllib.request.urlopen(req, context=self._ssl_context()) as resp:
                resp.read()
@@ -977,7 +1055,9 @@ class SiloClient:
            },
        )

-    def get_dag(self, part_number: str, revision_number: Optional[int] = None) -> Dict[str, Any]:
+    def get_dag(
+        self, part_number: str, revision_number: Optional[int] = None
+    ) -> Dict[str, Any]:
        """Fetch the stored DAG for an item.

        Calls ``GET /api/items/{partNumber}/dag``, optionally filtered
@@ -1052,7 +1132,9 @@ class SiloClient:

    def cancel_job(self, job_id: str) -> Dict[str, Any]:
        """Cancel a pending or running job."""
-        return self._request("POST", f"/jobs/{urllib.parse.quote(job_id, safe='')}/cancel")
+        return self._request(
+            "POST", f"/jobs/{urllib.parse.quote(job_id, safe='')}/cancel"
+        )

    # -- Job Definitions ----------------------------------------------------

@@ -1062,7 +1144,9 @@ class SiloClient:

    def get_job_definition(self, name: str) -> Dict[str, Any]:
        """Get a specific job definition by name."""
-        return self._request("GET", f"/job-definitions/{urllib.parse.quote(name, safe='')}")
+        return self._request(
+            "GET", f"/job-definitions/{urllib.parse.quote(name, safe='')}"
+        )

    def reload_job_definitions(self) -> Dict[str, Any]:
        """Re-read YAML job definitions from disk (admin only)."""
@@ -1080,7 +1164,9 @@ class SiloClient:

    def delete_runner(self, runner_id: str) -> None:
        """Delete a registered runner."""
-        self._request("DELETE", f"/runners/{urllib.parse.quote(runner_id, safe='')}", raw=True)
+        self._request(
+            "DELETE", f"/runners/{urllib.parse.quote(runner_id, safe='')}", raw=True
+        )

    # -- Runner-facing ------------------------------------------------------

@@ -1102,7 +1188,9 @@ class SiloClient:
            data["message"] = message
        return self._request("PUT", f"/runner/jobs/{jid}/progress", data)

-    def runner_complete_job(self, job_id: str, result: Optional[Dict] = None) -> Dict[str, Any]:
+    def runner_complete_job(
+        self, job_id: str, result: Optional[Dict] = None
+    ) -> Dict[str, Any]:
        """Mark a job as successfully completed."""
        jid = urllib.parse.quote(job_id, safe="")
        data: Dict[str, Any] = {}