kindred/silo-client
1
1
Fork 0
Code Issues 10 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add .kc metadata and dependency resolution API methods

Browse Source 
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
This commit is contained in:
Zoe Forbes
2026-02-18 19:35:09 -06:00
parent c5c8288eeb
commit f602eee7cc
1 changed files with 115 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
silo_client/__init__.py
View File

@@ -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] = {}