Buckets:
| from __future__ import annotations | |
| import hashlib | |
| import subprocess | |
| from pathlib import Path | |
| from typing import Any | |
| from observability.audit_log import utc_now | |
| def sha256_file(path: Path) -> str: | |
| digest = hashlib.sha256() | |
| with path.open("rb") as handle: | |
| for chunk in iter(lambda: handle.read(1024 * 1024), b""): | |
| digest.update(chunk) | |
| return f"sha256:{digest.hexdigest()}" | |
| def git_sha(cwd: Path) -> str: | |
| try: | |
| return subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=cwd, text=True, stderr=subprocess.DEVNULL).strip() | |
| except Exception: | |
| return "unknown" | |
| def base_manifest(manifest_type: str, run_id: str, payload: dict[str, Any]) -> dict[str, Any]: | |
| return { | |
| "manifest_type": manifest_type, | |
| "run_id": run_id, | |
| "created_at": utc_now(), | |
| **payload, | |
| } | |
Xet Storage Details
- Size:
- 855 Bytes
- Xet hash:
- cb3e2d895936c7bc34b5edbe3a102d072812888784c4c9cadc80a16374e357c1
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.