2 Commits

Author SHA1 Message Date
Zoe Forbes
fb658c5a24 feat: add push_dag and get_dag methods to SiloClient
- push_dag(part_number, revision_number, nodes, edges): PUT /api/items/{pn}/dag
- get_dag(part_number, revision_number=None): GET /api/items/{pn}/dag

Closes kindred/create#215
2026-02-14 15:06:22 -06:00
Zoe Forbes
68a4139251 fix: use _request() in delete_bom_entry() for consistent error handling (#59)
delete_bom_entry() used raw urllib.request instead of self._request(),
bypassing 401 auth clearing and standard error normalization. Replace
with a single _request('DELETE', ..., raw=True) call, matching the
pattern used by all other BOM methods.
2026-02-08 18:29:06 -06:00

View File

@@ -844,16 +844,7 @@ class SiloClient:
def delete_bom_entry(self, parent_pn: str, child_pn: str) -> None:
ppn = urllib.parse.quote(parent_pn, safe="")
cpn = urllib.parse.quote(child_pn, safe="")
url = f"{self.base_url}/items/{ppn}/bom/{cpn}"
headers = {"Content-Type": "application/json"}
headers.update(self._auth_headers())
req = urllib.request.Request(url, headers=headers, method="DELETE")
try:
urllib.request.urlopen(req, context=self._ssl_context())
except urllib.error.HTTPError as e:
raise RuntimeError(f"API error {e.code}: {e.read().decode()}")
except urllib.error.URLError as e:
raise RuntimeError(f"Connection error: {e.reason}")
self._request("DELETE", f"/items/{ppn}/bom/{cpn}", raw=True)
# -- Schemas ------------------------------------------------------------
@@ -911,3 +902,42 @@ class SiloClient:
self, ods_bytes: bytes, filename: str = "sheet.ods"
) -> Dict[str, Any]:
return self._upload_ods("/sheets/diff", ods_bytes, filename)
# -- DAG (feature tree) -------------------------------------------------
def push_dag(
self,
part_number: str,
revision_number: int,
nodes: List[Dict[str, Any]],
edges: List[Dict[str, Any]],
) -> Dict[str, Any]:
"""Push a feature DAG to the server.
Calls ``PUT /api/items/{partNumber}/dag`` with the payload
described in ``MULTI_USER_CLIENT.md`` Section 2.
"""
pn = urllib.parse.quote(part_number, safe="")
return self._request(
"PUT",
f"/items/{pn}/dag",
{
"revision_number": revision_number,
"nodes": nodes,
"edges": edges,
},
)
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
by *revision_number*.
"""
pn = urllib.parse.quote(part_number, safe="")
path = f"/items/{pn}/dag"
if revision_number is not None:
path += f"?revision={revision_number}"
return self._request("GET", path)