codekingpro/portable-devtools

1# Copyright 2025-present, the HuggingFace Inc. team.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#     http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14from dataclasses import dataclass15from datetime import datetime16from enum import Enum17from typing import Any18 19from huggingface_hub import constants20from huggingface_hub._space_api import SpaceHardware, Volume21from huggingface_hub.utils._datetime import parse_datetime22 23 24class JobStage(str, Enum):25    """26    Enumeration of possible stage of a Job on the Hub.27 28    Value can be compared to a string:29    ```py30    assert JobStage.COMPLETED == "COMPLETED"31    ```32    Possible values are: `COMPLETED`, `CANCELED`, `ERROR`, `DELETED`, `RUNNING`.33    Taken from https://github.com/huggingface/moon-landing/blob/main/server/job_types/JobInfo.ts#L61 (private url).34    """35 36    # Copied from moon-landing > server > lib > Job.ts37    COMPLETED = "COMPLETED"38    CANCELED = "CANCELED"39    ERROR = "ERROR"40    DELETED = "DELETED"41    RUNNING = "RUNNING"42 43 44@dataclass45class JobStatus:46    stage: JobStage47    message: str | None48 49 50@dataclass51class JobOwner:52    id: str53    name: str54    type: str55 56 57@dataclass58class JobInfo:59    """60    Contains information about a Job.61 62    Args:63        id (`str`):64            Job ID.65        created_at (`datetime` or `None`):66            When the Job was created.67        docker_image (`str` or `None`):68            The Docker image from Docker Hub used for the Job.69            Can be None if space_id is present instead.70        space_id (`str` or `None`):71            The Docker image from Hugging Face Spaces used for the Job.72            Can be None if docker_image is present instead.73        command (`list[str]` or `None`):74            Command of the Job, e.g. `["python", "-c", "print('hello world')"]`75        arguments (`list[str]` or `None`):76            Arguments passed to the command77        environment (`dict[str]` or `None`):78            Environment variables of the Job as a dictionary.79        secrets (`dict[str]` or `None`):80            Secret environment variables of the Job (encrypted).81        flavor (`str` or `None`):82            Flavor for the hardware, as in Hugging Face Spaces. See [`SpaceHardware`] for possible values.83            E.g. `"cpu-basic"`.84        labels (`dict[str, str]` or `None`):85            Labels to attach to the job (key-value pairs).86        volumes (`list[Volume]` or `None`):87            Volumes mounted in the job container (buckets, models, datasets, spaces).88        status: (`JobStatus` or `None`):89            Status of the Job, e.g. `JobStatus(stage="RUNNING", message=None)`90            See [`JobStage`] for possible stage values.91        owner: (`JobOwner` or `None`):92            Owner of the Job, e.g. `JobOwner(id="5e9ecfc04957053f60648a3e", name="lhoestq", type="user")`93 94    Example:95 96    ```python97    >>> from huggingface_hub import run_job98    >>> job = run_job(99    ...     image="python:3.12",100    ...     command=["python", "-c", "print('Hello from the cloud!')"]101    ... )102    >>> job103    JobInfo(id='687fb701029421ae5549d998', created_at=datetime.datetime(2025, 7, 22, 16, 6, 25, 79000, tzinfo=datetime.timezone.utc), docker_image='python:3.12', space_id=None, command=['python', '-c', "print('Hello from the cloud!')"], arguments=[], environment={}, secrets={}, flavor='cpu-basic', labels=None, status=JobStatus(stage='RUNNING', message=None), owner=JobOwner(id='5e9ecfc04957053f60648a3e', name='lhoestq', type='user'), endpoint='https://huggingface.co', url='https://huggingface.co/jobs/lhoestq/687fb701029421ae5549d998')104    >>> job.id105    '687fb701029421ae5549d998'106    >>> job.url107    'https://huggingface.co/jobs/lhoestq/687fb701029421ae5549d998'108    >>> job.status.stage109    'RUNNING'110    ```111    """112 113    id: str114    created_at: datetime | None115    docker_image: str | None116    space_id: str | None117    command: list[str] | None118    arguments: list[str] | None119    environment: dict[str, Any] | None120    secrets: dict[str, Any] | None121    flavor: SpaceHardware | None122    labels: dict[str, str] | None123    volumes: list[Volume] | None124    status: JobStatus125    owner: JobOwner126 127    # Inferred fields128    endpoint: str129    url: str130 131    def __init__(self, **kwargs) -> None:132        self.id = kwargs["id"]133        created_at = kwargs.get("createdAt") or kwargs.get("created_at")134        self.created_at = parse_datetime(created_at) if created_at else None135        self.docker_image = kwargs.get("dockerImage") or kwargs.get("docker_image")136        self.space_id = kwargs.get("spaceId") or kwargs.get("space_id")137        owner = kwargs.get("owner", {})138        self.owner = JobOwner(id=owner["id"], name=owner["name"], type=owner["type"])139        self.command = kwargs.get("command")140        self.arguments = kwargs.get("arguments")141        self.environment = kwargs.get("environment")142        self.secrets = kwargs.get("secrets")143        self.flavor = kwargs.get("flavor")144        self.labels = kwargs.get("labels")145        volumes = kwargs.get("volumes")146        self.volumes = [Volume(**v) for v in volumes] if volumes else None147        status = kwargs.get("status", {})148        self.status = JobStatus(stage=status["stage"], message=status.get("message"))149 150        # Inferred fields151        self.endpoint = kwargs.get("endpoint", constants.ENDPOINT)152        self.url = f"{self.endpoint}/jobs/{self.owner.name}/{self.id}"153 154 155@dataclass156class JobSpec:157    docker_image: str | None158    space_id: str | None159    command: list[str] | None160    arguments: list[str] | None161    environment: dict[str, Any] | None162    secrets: dict[str, Any] | None163    flavor: SpaceHardware | None164    timeout: int | None165    tags: list[str] | None166    arch: str | None167    labels: dict[str, str] | None168    volumes: list[Volume] | None169 170    def __init__(self, **kwargs) -> None:171        self.docker_image = kwargs.get("dockerImage") or kwargs.get("docker_image")172        self.space_id = kwargs.get("spaceId") or kwargs.get("space_id")173        self.command = kwargs.get("command")174        self.arguments = kwargs.get("arguments")175        self.environment = kwargs.get("environment")176        self.secrets = kwargs.get("secrets")177        self.flavor = kwargs.get("flavor")178        self.timeout = kwargs.get("timeout")179        self.tags = kwargs.get("tags")180        self.arch = kwargs.get("arch")181        self.labels = kwargs.get("labels")182        volumes = kwargs.get("volumes")183        self.volumes = [Volume(**v) for v in volumes] if volumes else None184 185 186@dataclass187class LastJobInfo:188    id: str189    at: datetime190 191    def __init__(self, **kwargs) -> None:192        self.id = kwargs["id"]193        self.at = parse_datetime(kwargs["at"])194 195 196@dataclass197class ScheduledJobStatus:198    last_job: LastJobInfo | None199    next_job_run_at: datetime | None200 201    def __init__(self, **kwargs) -> None:202        last_job = kwargs.get("lastJob") or kwargs.get("last_job")203        self.last_job = LastJobInfo(**last_job) if last_job else None204        next_job_run_at = kwargs.get("nextJobRunAt") or kwargs.get("next_job_run_at")205        self.next_job_run_at = parse_datetime(str(next_job_run_at)) if next_job_run_at else None206 207 208@dataclass209class ScheduledJobInfo:210    """211    Contains information about a Job.212 213    Args:214        id (`str`):215            Scheduled Job ID.216        created_at (`datetime` or `None`):217            When the scheduled Job was created.218        tags (`list[str]` or `None`):219            The tags of the scheduled Job.220        schedule (`str` or `None`):221            One of "@annually", "@yearly", "@monthly", "@weekly", "@daily", "@hourly", or a222            CRON schedule expression (e.g., '0 9 * * 1' for 9 AM every Monday).223        suspend (`bool` or `None`):224            Whether the scheduled job is suspended (paused).225        concurrency (`bool` or `None`):226            Whether multiple instances of this Job can run concurrently.227        status (`ScheduledJobStatus` or `None`):228            Status of the scheduled Job.229        owner: (`JobOwner` or `None`):230            Owner of the scheduled Job, e.g. `JobOwner(id="5e9ecfc04957053f60648a3e", name="lhoestq", type="user")`231        job_spec: (`JobSpec` or `None`):232            Specifications of the Job.233 234    Example:235 236    ```python237    >>> from huggingface_hub import run_job238    >>> scheduled_job = create_scheduled_job(239    ...     image="python:3.12",240    ...     command=["python", "-c", "print('Hello from the cloud!')"],241    ...     schedule="@hourly",242    ... )243    >>> scheduled_job.id244    '687fb701029421ae5549d999'245    >>> scheduled_job.status.next_job_run_at246    datetime.datetime(2025, 7, 22, 17, 6, 25, 79000, tzinfo=datetime.timezone.utc)247    ```248    """249 250    id: str251    created_at: datetime | None252    job_spec: JobSpec253    schedule: str | None254    suspend: bool | None255    concurrency: bool | None256    status: ScheduledJobStatus257    owner: JobOwner258 259    def __init__(self, **kwargs) -> None:260        self.id = kwargs["id"]261        created_at = kwargs.get("createdAt") or kwargs.get("created_at")262        self.created_at = parse_datetime(created_at) if created_at else None263        self.job_spec = JobSpec(**(kwargs.get("job_spec") or kwargs.get("jobSpec", {})))264        self.schedule = kwargs.get("schedule")265        self.suspend = kwargs.get("suspend")266        self.concurrency = kwargs.get("concurrency")267        status = kwargs.get("status", {})268        self.status = ScheduledJobStatus(269            last_job=status.get("last_job") or status.get("lastJob"),270            next_job_run_at=status.get("next_job_run_at") or status.get("nextJobRunAt"),271        )272        owner = kwargs.get("owner", {})273        self.owner = JobOwner(id=owner["id"], name=owner["name"], type=owner["type"])274 275 276@dataclass277class JobAccelerator:278    """279    Contains information about a Job accelerator (GPU).280 281    Args:282        type (`str`):283            Type of accelerator, e.g. `"gpu"`.284        model (`str`):285            Model of accelerator, e.g. `"T4"`, `"A10G"`, `"A100"`, `"L4"`, `"L40S"`.286        quantity (`str`):287            Number of accelerators, e.g. `"1"`, `"2"`, `"4"`, `"8"`.288        vram (`str`):289            Total VRAM, e.g. `"16 GB"`, `"24 GB"`.290        manufacturer (`str`):291            Manufacturer of the accelerator, e.g. `"Nvidia"`.292    """293 294    type: str295    model: str296    quantity: str297    vram: str298    manufacturer: str299 300    def __init__(self, **kwargs) -> None:301        self.type = kwargs["type"]302        self.model = kwargs["model"]303        self.quantity = kwargs["quantity"]304        self.vram = kwargs["vram"]305        self.manufacturer = kwargs["manufacturer"]306 307 308@dataclass309class JobHardware:310    """311    Contains information about available Job hardware.312 313    Args:314        name (`str`):315            Machine identifier, e.g. `"cpu-basic"`, `"a10g-large"`.316        pretty_name (`str`):317            Human-readable name, e.g. `"CPU Basic"`, `"Nvidia A10G - large"`.318        cpu (`str`):319            CPU specification, e.g. `"2 vCPU"`, `"12 vCPU"`.320        ram (`str`):321            RAM specification, e.g. `"16 GB"`, `"46 GB"`.322        accelerator (`JobAccelerator` or `None`):323            GPU/accelerator details if available.324        unit_cost_micro_usd (`int`):325            Cost in micro-dollars per unit, e.g. `167` (= $0.000167).326        unit_cost_usd (`float`):327            Cost in USD per unit, e.g. `0.000167`.328        unit_label (`str`):329            Cost unit period, e.g. `"minute"`.330 331    Example:332 333    ```python334    >>> from huggingface_hub import list_jobs_hardware335    >>> hardware_list = list_jobs_hardware()336    >>> hardware_list[0]337    JobHardware(name='cpu-basic', pretty_name='CPU Basic', cpu='2 vCPU', ram='16 GB', accelerator=None, unit_cost_micro_usd=167, unit_cost_usd=0.000167, unit_label='minute')338    >>> hardware_list[0].name339    'cpu-basic'340    ```341    """342 343    name: str344    pretty_name: str345    cpu: str346    ram: str347    accelerator: JobAccelerator | None348    unit_cost_micro_usd: int349    unit_cost_usd: float350    unit_label: str351 352    def __init__(self, **kwargs) -> None:353        self.name = kwargs["name"]354        self.pretty_name = kwargs["prettyName"]355        self.cpu = kwargs["cpu"]356        self.ram = kwargs["ram"]357        accelerator = kwargs.get("accelerator")358        self.accelerator = JobAccelerator(**accelerator) if accelerator else None359        self.unit_cost_micro_usd = kwargs["unitCostMicroUSD"]360        self.unit_cost_usd = kwargs["unitCostUSD"]361        self.unit_label = kwargs["unitLabel"]362 363 364def _create_job_spec(365    *,366    image: str,367    command: list[str],368    env: dict[str, Any] | None,369    secrets: dict[str, Any] | None,370    flavor: SpaceHardware | None,371    timeout: int | float | str | None,372    labels: dict[str, str] | None = None,373    volumes: list[Volume] | None = None,374) -> dict[str, Any]:375    # prepare job spec to send to HF Jobs API376    job_spec: dict[str, Any] = {377        "command": command,378        "arguments": [],379        "environment": env or {},380        "flavor": flavor or SpaceHardware.CPU_BASIC,381    }382    # secrets are optional383    if secrets:384        job_spec["secrets"] = secrets385    # timeout is optional386    if timeout:387        time_units_factors = {"s": 1, "m": 60, "h": 3600, "d": 3600 * 24}388        if isinstance(timeout, str) and timeout[-1] in time_units_factors:389            job_spec["timeoutSeconds"] = int(float(timeout[:-1]) * time_units_factors[timeout[-1]])390        else:391            job_spec["timeoutSeconds"] = int(timeout)392    # labels are optional393    if labels:394        job_spec["labels"] = labels395    # volumes are optional396    if volumes:397        job_spec["volumes"] = [vol.to_dict() for vol in volumes]398    # input is either from docker hub or from HF spaces399    for prefix in (400        "https://huggingface.co/spaces/",401        "https://hf.co/spaces/",402        "huggingface.co/spaces/",403        "hf.co/spaces/",404    ):405        if image.startswith(prefix):406            job_spec["spaceId"] = image[len(prefix) :]407            break408    else:409        job_spec["dockerImage"] = image410    return job_spec411