"""Minimal IPFS client — upload + pin via kubo's HTTP API. Designed for the ``example-studio`` shape: a small local kubo daemon runs alongside the web app, accepts uploads from end users (e.g. a customer uploading a photo of a finished custom order), pins locally for fast serving, and optionally mirrors pins to a second remote node (the cold host-on-LAN) for archival redundancy. No IPFS libraries are imported — just httpx against the kubo REST API (v0). Keeps the SDK surface minimal. """ from __future__ import annotations import logging from dataclasses import dataclass from typing import Optional import httpx logger = logging.getLogger(__name__) @dataclass class IPFSClient: """Kubo-compatible IPFS client. Attributes: api_url: Base URL of the kubo HTTP API (default ``http://127.0.0.1:5001``). timeout: Per-request timeout in seconds (default 60 — uploads can be slow). mirror_api_urls: Optional list of additional kubo endpoints to ``pin add`` the CID on after a successful primary pin. Use this to mirror to the cold host or any other archival node. """ api_url: str = "http://127.0.0.1:5001" timeout: float = 60.0 mirror_api_urls: list[str] = None # type: ignore[assignment] def __post_init__(self) -> None: if self.mirror_api_urls is None: self.mirror_api_urls = [] async def add(self, data: bytes, filename: str = "upload") -> str: """Upload bytes and pin them locally. Args: data: Raw bytes to add. filename: Logical name used by clients browsing the DAG (doesn't affect the CID). Returns: CID (base58, v0 or base32 v1 depending on kubo defaults). Raises: RuntimeError: If the daemon is unreachable or returns a non-2xx. """ url = f"{self.api_url.rstrip('/')}/api/v0/add" async with httpx.AsyncClient(timeout=self.timeout) as client: resp = await client.post( url, files={"file": (filename, data, "application/octet-stream")}, params={"pin": "true", "cid-version": "1"}, ) if resp.status_code >= 400: raise RuntimeError(f"ipfs add {resp.status_code}: {resp.text[:200]}") # kubo's /add streams NDJSON; each line is one {Name, Hash, Size}. # For a single file upload the last line carries the wrapping CID. last_cid: Optional[str] = None for line in resp.text.strip().splitlines(): if '"Hash"' in line: import json obj = json.loads(line) last_cid = obj.get("Hash") if not last_cid: raise RuntimeError(f"ipfs add: no CID in response: {resp.text[:200]}") # Mirror pins (best effort — a mirror failure should not poison the primary upload). for mirror in self.mirror_api_urls: try: await self._pin_on(mirror, last_cid) except Exception as exc: logger.warning("[ipfs] mirror pin to %s failed for %s: %s", mirror, last_cid, exc) return last_cid async def _pin_on(self, api_url: str, cid: str) -> None: """Pin an existing CID on a remote kubo node.""" url = f"{api_url.rstrip('/')}/api/v0/pin/add" async with httpx.AsyncClient(timeout=self.timeout) as client: resp = await client.post(url, params={"arg": cid}) if resp.status_code >= 400: raise RuntimeError(f"pin/add {resp.status_code}: {resp.text[:200]}") async def pin_bytes( data: bytes, api_url: str = "http://127.0.0.1:5001", mirror_api_urls: Optional[list[str]] = None, filename: str = "upload", ) -> str: """Convenience wrapper: one-shot upload + pin (+ optional mirror). Returns the CID. """ client = IPFSClient(api_url=api_url, mirror_api_urls=mirror_api_urls or []) return await client.add(data, filename=filename)