File size: 1,429 Bytes
6e53100 | 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 | from __future__ import annotations
import os
from dataclasses import dataclass
@dataclass(frozen=True)
class Settings:
"""Runtime configuration for the orchestrator Space."""
bucket_name: str = os.getenv("SPACE_FACTORY_BUCKET_NAME", "space-factory-runs")
bucket_mount: str = os.getenv("SPACE_FACTORY_BUCKET_MOUNT", "/output")
job_flavor: str = os.getenv("SPACE_FACTORY_JOB_FLAVOR", "cpu-basic")
job_timeout: str = os.getenv("SPACE_FACTORY_JOB_TIMEOUT", "30m")
job_image: str = os.getenv("SPACE_FACTORY_JOB_IMAGE", "python:3.12")
def normalize_bucket_name(value: str | None) -> str:
"""Return a safe bucket name, defaulting to the product bucket name."""
bucket = (value or settings.bucket_name).strip().strip("/")
if not bucket:
bucket = settings.bucket_name
if "/" in bucket:
raise ValueError("Bucket name must be a name like 'space-factory-runs', not owner/name.")
return bucket
def user_bucket_source(*, username: str, bucket_name: str | None = None) -> str:
"""Return the per-user bucket source expected by HF Jobs volumes."""
clean_user = (username or "").strip()
if not clean_user:
raise ValueError("Missing username for per-user bucket source.")
return f"{clean_user}/{normalize_bucket_name(bucket_name)}"
def bucket_uri_from_source(bucket_source: str) -> str:
return f"hf://buckets/{bucket_source}"
settings = Settings()
|