File size: 3,049 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 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | """Filecoin-pinned IPFS via Lighthouse Storage.
Direct httpx calls to the Lighthouse REST API — no SDK dependency. Returns a
content-addressed `cid://...` URI suitable for the provenance manifest's
`lighthouse_cid` field.
Falls back to a deterministic local-hash CID stub when `LIGHTHOUSE_API_KEY`
is unset (so dev/CI runs don't need credentials).
"""
from __future__ import annotations
import os
import tarfile
import tempfile
from pathlib import Path
import httpx
from mindxtrain.provenance.hashing import blake3_dir
from mindxtrain.storage.provider import StorageProvider, StorageRef
def _stub_cid(checkpoint_dir: Path) -> str:
digest = blake3_dir(checkpoint_dir)
return f"cid://stub-{digest[:32]}"
def publish_to_lighthouse(
checkpoint_dir: Path,
*,
api_key: str | None = None,
base_url: str | None = None,
timeout_s: float = 600.0,
) -> str:
"""Tar + pin `checkpoint_dir` to Lighthouse; return a `cid://...` URI.
If `LIGHTHOUSE_API_KEY` is unset we return a deterministic stub CID
derived from BLAKE3, so dev runs still produce a valid manifest.
"""
api_key = api_key or os.environ.get("LIGHTHOUSE_API_KEY", "")
if not api_key:
return _stub_cid(checkpoint_dir)
base_url = (base_url or os.environ.get("LIGHTHOUSE_BASE_URL", "https://node.lighthouse.storage")).rstrip("/")
# Tar the directory to a temp file so we upload one binary blob.
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(checkpoint_dir), arcname=checkpoint_dir.name)
with tar_path.open("rb") as fh, httpx.Client(timeout=timeout_s) as client:
resp = client.post(
f"{base_url}/api/v0/add",
files={"file": (f"{checkpoint_dir.name}.tar", fh, "application/x-tar")},
headers={"Authorization": f"Bearer {api_key}"},
)
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"Lighthouse response missing CID: {data}"
raise RuntimeError(msg)
return f"cid://{cid}"
class LighthouseProvider(StorageProvider):
"""`StorageProvider` adapter over `publish_to_lighthouse`."""
name = "lighthouse"
def put_dir(self, src: Path, key: str) -> StorageRef:
_ = key
cid = publish_to_lighthouse(src)
return StorageRef(provider=self.name, uri=cid)
def get_dir(self, ref: StorageRef, dest: Path) -> Path:
# Lighthouse fetch goes through any IPFS gateway; not implemented here.
# The user can `ipfs get <cid>` from a kubo node or use the
# `mindxtrain.storage.ipfs` provider for downloads.
msg = "LighthouseProvider.get_dir — fetch via mindxtrain.storage.ipfs or kubo CLI."
raise NotImplementedError(msg)
__all__ = ["LighthouseProvider", "publish_to_lighthouse"]
|