Add Bambuddy InvenTree sync service

src/bambuddy_inventree_sync/inventree.py

@@ -0,0 +1,103 @@
+from typing import Any
+
+import httpx
+
+from .config import Settings
+from .http_errors import ExternalApiError
+from .models import Archive
+
+
+class InvenTreeClient:
+    def __init__(self, settings: Settings) -> None:
+        base_url = settings.inventree_base_url.rstrip("/")
+        if not base_url.endswith("/api"):
+            base_url = f"{base_url}/api"
+
+        self.settings = settings
+        self.base_url = base_url
+        self.client = httpx.AsyncClient(
+            base_url=self.base_url,
+            headers={
+                "Authorization": f"Token {settings.inventree_token}",
+                "Accept": "application/json",
+                "Content-Type": "application/json",
+            },
+            timeout=settings.http_timeout_seconds,
+            trust_env=False,
+        )
+
+    async def close(self) -> None:
+        await self.client.aclose()
+
+    async def get_part_category(self) -> dict[str, Any]:
+        return await self._request("GET", f"/part/category/{self.settings.inventree_part_category_id}/")
+
+    async def get_stock_location(self) -> dict[str, Any]:
+        return await self._request("GET", f"/stock/location/{self.settings.inventree_stock_location_id}/")
+
+    async def find_part_by_ipn(self, ipn: str) -> dict[str, Any] | None:
+        for params in ({"IPN": ipn, "limit": 100}, {"search": ipn, "limit": 100}):
+            data = await self._request("GET", "/part/", params=params)
+            for item in self._items(data):
+                if item.get("IPN") == ipn:
+                    return item
+        return None
+
+    async def create_part(self, *, name: str, description: str, ipn: str) -> dict[str, Any]:
+        payload = {
+            "name": name[:100],
+            "description": description,
+            "category": self.settings.inventree_part_category_id,
+            "IPN": ipn[:100],
+            "active": True,
+            "component": True,
+            "purchaseable": False,
+            "salable": False,
+            "assembly": False,
+            "trackable": False,
+            "virtual": False,
+        }
+        return await self._request("POST", "/part/", json=payload)
+
+    async def find_stock_by_batch(self, *, part_id: int, batch: str) -> dict[str, Any] | None:
+        data = await self._request("GET", "/stock/", params={"part": part_id, "batch": batch, "limit": 100})
+        for item in self._items(data):
+            if item.get("part") == part_id and item.get("batch") == batch:
+                return item
+        return None
+
+    async def create_stock_item(self, *, part_id: int, archive: Archive, notes: str) -> dict[str, Any]:
+        quantity = archive.quantity or self.settings.default_stock_quantity
+        payload: dict[str, Any] = {
+            "part": part_id,
+            "location": self.settings.inventree_stock_location_id,
+            "quantity": quantity,
+            "status": self.settings.inventree_stock_status,
+            "batch": self.batch_for_archive(archive.id),
+            "notes": notes,
+        }
+        return await self._request("POST", "/stock/", json=payload)
+
+    @staticmethod
+    def batch_for_archive(archive_id: int) -> str:
+        return f"bambuddy-{archive_id}"
+
+    async def _request(self, method: str, path: str, **kwargs: Any) -> Any:
+        response = await self.client.request(method, path, **kwargs)
+        if response.status_code >= 400:
+            body = response.text[:1000]
+            raise ExternalApiError(f"InvenTree {method} {path} failed: HTTP {response.status_code}: {body}")
+
+        if response.content:
+            return response.json()
+        return None
+
+    @staticmethod
+    def _items(data: Any) -> list[dict[str, Any]]:
+        if isinstance(data, list):
+            return data
+        if isinstance(data, dict):
+            results = data.get("results")
+            if isinstance(results, list):
+                return results
+        return []