File size: 2,716 Bytes
dfb775d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | """Raw IPFS `StorageProvider` via kubo HTTP API.
For users running their own kubo node (`ipfs daemon`) rather than going
through Lighthouse/Filecoin. Direct httpx — no SDK dependency.
"""
from __future__ import annotations
import os
import tarfile
import tempfile
from pathlib import Path
import httpx
from mindxtrain.storage.provider import StorageProvider, StorageRef
class IpfsProvider(StorageProvider):
name = "ipfs"
def __init__(self, api_url: str | None = None, timeout_s: float = 600.0) -> None:
self.api_url = (api_url or os.environ.get("IPFS_API_URL", "http://127.0.0.1:5001")).rstrip("/")
self.timeout_s = timeout_s
def put_dir(self, src: Path, key: str) -> StorageRef:
"""Tar + add the directory; return a `cid://<hash>` ref."""
_ = key
with tempfile.NamedTemporaryFile(suffix=".tar", delete=False) as tmp:
tar_path = Path(tmp.name)
try:
with tarfile.open(tar_path, "w") as tf:
tf.add(str(src), arcname=src.name)
with tar_path.open("rb") as fh, httpx.Client(timeout=self.timeout_s) as client:
resp = client.post(
f"{self.api_url}/api/v0/add",
files={"file": (f"{src.name}.tar", fh, "application/x-tar")},
)
resp.raise_for_status()
data = resp.json()
finally:
tar_path.unlink(missing_ok=True)
cid = data.get("Hash") or data.get("cid")
if not cid:
msg = f"kubo /api/v0/add missing Hash: {data}"
raise RuntimeError(msg)
return StorageRef(provider=self.name, uri=f"cid://{cid}")
def get_dir(self, ref: StorageRef, dest: Path) -> Path:
"""Pull the tar via /api/v0/cat and unpack into dest."""
if ref.provider != self.name:
msg = f"ref provider {ref.provider!r} != {self.name!r}"
raise ValueError(msg)
cid = ref.uri.removeprefix("cid://")
dest.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(suffix=".tar", delete=False) as tmp:
tar_path = Path(tmp.name)
try:
with httpx.Client(timeout=self.timeout_s) as client:
with client.stream("POST", f"{self.api_url}/api/v0/cat?arg={cid}") as resp:
resp.raise_for_status()
with tar_path.open("wb") as fh:
for chunk in resp.iter_bytes():
fh.write(chunk)
with tarfile.open(tar_path) as tf:
tf.extractall(dest)
finally:
tar_path.unlink(missing_ok=True)
return dest
__all__ = ["IpfsProvider"]
|