venv/lib/python3.11/site-packages/wandb/sdk/launch/create_job.py

import json
import logging
import os
import re
import sys
import tempfile
from typing import Any, Dict, List, Optional, Tuple

import wandb
from wandb.apis.internal import Api
from wandb.sdk.artifacts.artifact import Artifact
from wandb.sdk.internal.job_builder import JobBuilder
from wandb.sdk.launch.git_reference import GitReference
from wandb.sdk.launch.utils import (
    _is_git_uri,
    get_current_python_version,
    get_entrypoint_file,
)
from wandb.sdk.lib import filesystem
from wandb.util import make_artifact_name_safe

logging.basicConfig(stream=sys.stdout, level=logging.INFO)
_logger = logging.getLogger("wandb")


CODE_ARTIFACT_EXCLUDE_PATHS = ["wandb", ".git"]


def create_job(
    path: str,
    job_type: str,
    entity: Optional[str] = None,
    project: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    aliases: Optional[List[str]] = None,
    runtime: Optional[str] = None,
    entrypoint: Optional[str] = None,
    git_hash: Optional[str] = None,
    build_context: Optional[str] = None,
    dockerfile: Optional[str] = None,
) -> Optional[Artifact]:
    """Create a job from a path, not as the output of a run.

    Arguments:
        path (str): Path to the job directory.
        job_type (str): Type of the job. One of "git", "code", or "image".
        entity (Optional[str]): Entity to create the job under.
        project (Optional[str]): Project to create the job under.
        name (Optional[str]): Name of the job.
        description (Optional[str]): Description of the job.
        aliases (Optional[List[str]]): Aliases for the job.
        runtime (Optional[str]): Python runtime of the job, like 3.9.
        entrypoint (Optional[str]): Entrypoint of the job. If build_context is
            provided, path is relative to build_context.
        git_hash (Optional[str]): Git hash of a specific commit, when using git type jobs.
        build_context (Optional[str]): Path to the build context, when using image type jobs.
        dockerfile (Optional[str]): Path to the Dockerfile, when using image type jobs.
            If build_context is provided, path is relative to build_context.

    Returns:
        Optional[Artifact]: The artifact created by the job, the action (for printing), and job aliases.
                            None if job creation failed.

    Example:
        ```python
        artifact_job = wandb.create_job(
            job_type="code",
            path=".",
            entity="wandb",
            project="jobs",
            name="my-train-job",
            description="My training job",
            aliases=["train"],
            runtime="3.9",
            entrypoint="train.py",
        )
        # then run the newly created job
        artifact_job.call()
        ```
    """
    api = Api()

    artifact_job, _action, _aliases = _create_job(
        api,
        job_type,
        path,
        entity,
        project,
        name,
        description,
        aliases,
        runtime,
        entrypoint,
        git_hash,
        build_context,
        dockerfile,
    )

    return artifact_job


def _create_job(
    api: Api,
    job_type: str,
    path: str,
    entity: Optional[str] = None,
    project: Optional[str] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    aliases: Optional[List[str]] = None,
    runtime: Optional[str] = None,
    entrypoint: Optional[str] = None,
    git_hash: Optional[str] = None,
    build_context: Optional[str] = None,
    dockerfile: Optional[str] = None,
    base_image: Optional[str] = None,
) -> Tuple[Optional[Artifact], str, List[str]]:
    wandb.termlog(f"Creating launch job of type: {job_type}...")

    if name and name != make_artifact_name_safe(name):
        wandb.termerror(
            f"Artifact names may only contain alphanumeric characters, dashes, underscores, and dots. Did you mean: {make_artifact_name_safe(name)}"
        )
        return None, "", []

    if runtime is not None:
        if not re.match(r"^3\.\d+$", runtime):
            wandb.termerror(
                f"Runtime (-r, --runtime) must be a minor version of Python 3, "
                f"e.g. 3.9 or 3.10, received {runtime}"
            )
            return None, "", []
    aliases = aliases or []
    tempdir = tempfile.TemporaryDirectory()
    try:
        metadata, requirements = _make_metadata_for_partial_job(
            job_type=job_type,
            tempdir=tempdir,
            git_hash=git_hash,
            runtime=runtime,
            path=path,
            entrypoint=entrypoint,
        )
        if not metadata:
            return None, "", []
    except Exception as e:
        wandb.termerror(f"Error creating job: {e}")
        return None, "", []

    _dump_metadata_and_requirements(
        metadata=metadata,
        tmp_path=tempdir.name,
        requirements=requirements,
    )

    try:
        # init hidden wandb run with job building disabled (handled manually)
        run = wandb.init(
            dir=tempdir.name,
            settings={"silent": True, "disable_job_creation": True},
            entity=entity,
            project=project,
            job_type="cli_create_job",
        )
    except Exception:
        # Error printed by wandb.init
        return None, "", []

    job_builder = _configure_job_builder_for_partial(tempdir.name, job_source=job_type)
    job_builder._settings.job_name = name
    if job_type == "code":
        assert entrypoint is not None
        job_name = _make_code_artifact(
            api=api,
            job_builder=job_builder,
            path=path,
            entrypoint=entrypoint,
            run=run,  # type: ignore
            entity=entity,
            project=project,
            name=name,
        )
        if not job_name:
            return None, "", []
        name = job_name

    # build job artifact, loads wandb-metadata and creates wandb-job.json here
    artifact = job_builder.build(
        api.api,
        dockerfile=dockerfile,
        build_context=build_context,
        base_image=base_image,
    )
    if not artifact:
        wandb.termerror("JobBuilder failed to build a job")
        _logger.debug("Failed to build job, check job source and metadata")
        return None, "", []

    if not name:
        name = artifact.name

    aliases += job_builder._aliases
    if "latest" not in aliases:
        aliases += ["latest"]

    res, _ = api.create_artifact(
        artifact_type_name="job",
        artifact_collection_name=name,
        digest=artifact.digest,
        client_id=artifact._client_id,
        sequence_client_id=artifact._sequence_client_id,
        entity_name=entity,
        project_name=project,
        run_name=run.id,  # type: ignore # run will be deleted after creation
        description=description,
        metadata={"_partial": True},
        is_user_created=True,
        aliases=[{"artifactCollectionName": name, "alias": a} for a in aliases],
    )
    action = "No changes detected for"
    if not res.get("artifactSequence", {}).get("latestArtifact"):
        # When there is no latestArtifact, we are creating new
        action = "Created"
    elif res.get("state") == "PENDING":
        # updating an existing artifafct, state is pending awaiting call to
        # log_artifact to upload and finalize artifact. If not pending, digest
        # is the same as latestArtifact, so no changes detected
        action = "Updated"

    run.log_artifact(artifact, aliases=aliases)  # type: ignore
    artifact.wait()
    run.finish()  # type: ignore

    # fetch, then delete hidden run
    _run = wandb.Api().run(f"{entity}/{project}/{run.id}")  # type: ignore
    _run.delete()

    return artifact, action, aliases


def _make_metadata_for_partial_job(
    job_type: str,
    tempdir: tempfile.TemporaryDirectory,
    git_hash: Optional[str],
    runtime: Optional[str],
    path: str,
    entrypoint: Optional[str],
) -> Tuple[Optional[Dict[str, Any]], Optional[List[str]]]:
    """Create metadata for partial jobs, return metadata and requirements."""
    metadata = {}
    if job_type == "git":
        assert entrypoint is not None
        repo_metadata = _create_repo_metadata(
            path=path,
            tempdir=tempdir.name,
            entrypoint=entrypoint,
            git_hash=git_hash,
            runtime=runtime,
        )
        if not repo_metadata:
            tempdir.cleanup()  # otherwise git can pollute
            return None, None
        metadata.update(repo_metadata)
        return metadata, None

    if job_type == "code":
        assert entrypoint is not None
        artifact_metadata, requirements = _create_artifact_metadata(
            path=path, entrypoint=entrypoint, runtime=runtime
        )
        if not artifact_metadata:
            return None, None
        metadata.update(artifact_metadata)
        return metadata, requirements

    if job_type == "image":
        if runtime:
            wandb.termwarn(
                "Setting runtime is not supported for image jobs, ignoring runtime"
            )
        # TODO(gst): support entrypoint for image based jobs
        if entrypoint:
            wandb.termwarn(
                "Setting an entrypoint is not currently supported for image jobs, ignoring entrypoint argument"
            )
        metadata.update({"python": runtime or "", "docker": path})
        return metadata, None

    wandb.termerror(f"Invalid job type: {job_type}")
    return None, None


def _maybe_warn_python_no_executable(entrypoint: str):
    entrypoint_list = entrypoint.split(" ")
    if len(entrypoint_list) == 1 and entrypoint_list[0].endswith(".py"):
        wandb.termwarn(
            f"Entrypoint {entrypoint} is a python file without an executable, you may want to use `python {entrypoint}` as the entrypoint instead."
        )


def _create_repo_metadata(
    path: str,
    tempdir: str,
    entrypoint: str,
    git_hash: Optional[str] = None,
    runtime: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
    # Make sure the entrypoint doesn't contain any backward path traversal
    if entrypoint and ".." in entrypoint:
        wandb.termerror("Entrypoint cannot contain backward path traversal")
        return None

    _maybe_warn_python_no_executable(entrypoint)

    if not _is_git_uri(path):
        wandb.termerror("Path must be a git URI")
        return None

    ref = GitReference(path, git_hash)
    if not ref:
        wandb.termerror("Could not parse git URI")
        return None

    ref.fetch(tempdir)

    commit = ref.commit_hash
    if not commit:
        if not ref.commit_hash:
            wandb.termerror("Could not find git commit hash")
            return None
        commit = ref.commit_hash

    local_dir = os.path.join(tempdir, ref.path or "")
    python_version = runtime
    if not python_version:
        if os.path.exists(os.path.join(local_dir, "runtime.txt")):
            with open(os.path.join(local_dir, "runtime.txt")) as f:
                python_version = f.read().strip()
        elif os.path.exists(os.path.join(local_dir, ".python-version")):
            with open(os.path.join(local_dir, ".python-version")) as f:
                python_version = f.read().strip().splitlines()[0]
        else:
            python_version, _ = get_current_python_version()

    python_version = _clean_python_version(python_version)

    metadata = {
        "git": {
            "commit": commit,
            "remote": ref.url,
        },
        "entrypoint": entrypoint.split(" "),
        "python": python_version,  # used to build container
        "notebook": False,  # partial jobs from notebooks not supported
    }

    return metadata


def _create_artifact_metadata(
    path: str, entrypoint: str, runtime: Optional[str] = None
) -> Tuple[Optional[Dict[str, Any]], Optional[List[str]]]:
    if not os.path.isdir(path):
        wandb.termerror("Path must be a valid file or directory")
        return {}, []

    _maybe_warn_python_no_executable(entrypoint)

    entrypoint_list = entrypoint.split(" ")
    entrypoint_file = get_entrypoint_file(entrypoint_list)

    # read local requirements.txt and dump to temp dir for builder
    requirements = []
    depspath = os.path.join(path, "requirements.txt")
    if os.path.exists(depspath):
        with open(depspath) as f:
            requirements = f.read().splitlines()

    if not any(["wandb" in r for r in requirements]):
        wandb.termwarn("wandb is not present in requirements.txt.")

    if runtime:
        python_version = _clean_python_version(runtime)
    else:
        python_version, _ = get_current_python_version()
        python_version = _clean_python_version(python_version)

    metadata = {
        "python": python_version,
        "codePath": entrypoint_file,
        "entrypoint": entrypoint_list,
    }
    return metadata, requirements


def _configure_job_builder_for_partial(tmpdir: str, job_source: str) -> JobBuilder:
    """Configure job builder with temp dir and job source."""
    # adjust git source to repo
    if job_source == "git":
        job_source = "repo"

    # adjust code source to artifact
    if job_source == "code":
        job_source = "artifact"

    settings = wandb.Settings(x_files_dir=tmpdir, job_source=job_source)
    job_builder = JobBuilder(
        settings=settings,  # type: ignore
        verbose=True,
    )
    job_builder._partial = True
    # never allow notebook runs
    job_builder._is_notebook_run = False
    # set run inputs and outputs to empty dicts
    job_builder.set_config({})
    job_builder.set_summary({})
    return job_builder


def _make_code_artifact(
    api: Api,
    job_builder: JobBuilder,
    run: "wandb.sdk.wandb_run.Run",
    path: str,
    entrypoint: str,
    entity: Optional[str],
    project: Optional[str],
    name: Optional[str],
) -> Optional[str]:
    """Helper for creating and logging code artifacts.

    Returns the name of the eventual job.
    """
    entrypoint_list = entrypoint.split(" ")
    # We no longer require the entrypoint to end in an existing file. But we
    # need something to use as the default job artifact name. In the future we
    # may require the user to provide a job name explicitly when calling
    # wandb job create.
    entrypoint_file = entrypoint_list[-1]
    artifact_name = _make_code_artifact_name(os.path.join(path, entrypoint_file), name)
    code_artifact = wandb.Artifact(
        name=artifact_name,
        type="code",
        description="Code artifact for job",
    )

    try:
        code_artifact.add_dir(path)
    except Exception as e:
        if os.path.islink(path):
            wandb.termerror(
                "Symlinks are not supported for code artifact jobs, please copy the code into a directory and try again"
            )
        wandb.termerror(f"Error adding to code artifact: {e}")
        return None

    # Remove paths we don't want to include, if present
    for item in CODE_ARTIFACT_EXCLUDE_PATHS:
        try:
            code_artifact.remove(item)
        except FileNotFoundError:
            pass

    res, _ = api.create_artifact(
        artifact_type_name="code",
        artifact_collection_name=artifact_name,
        digest=code_artifact.digest,
        client_id=code_artifact._client_id,
        sequence_client_id=code_artifact._sequence_client_id,
        entity_name=entity,
        project_name=project,
        run_name=run.id,  # run will be deleted after creation
        description="Code artifact for job",
        metadata={"codePath": path, "entrypoint": entrypoint_file},
        is_user_created=True,
        aliases=[
            {"artifactCollectionName": artifact_name, "alias": a} for a in ["latest"]
        ],
    )
    run.log_artifact(code_artifact)
    code_artifact.wait()
    job_builder._handle_server_artifact(res, code_artifact)  # type: ignore

    # code artifacts have "code" prefix, remove it and alias
    if not name:
        name = code_artifact.name.replace("code", "job").split(":")[0]

    return name


def _make_code_artifact_name(path: str, name: Optional[str]) -> str:
    """Make a code artifact name from a path and user provided name."""
    if name:
        return f"code-{name}"

    clean_path = path.replace("./", "")
    if clean_path[0] == "/":
        clean_path = clean_path[1:]
    if clean_path[-1] == "/":
        clean_path = clean_path[:-1]

    path_name = f"code-{make_artifact_name_safe(clean_path)}"
    return path_name


def _dump_metadata_and_requirements(
    tmp_path: str, metadata: Dict[str, Any], requirements: Optional[List[str]]
) -> None:
    """Dump manufactured metadata and requirements.txt.

    File used by the job_builder to create a job from provided metadata.
    """
    filesystem.mkdir_exists_ok(tmp_path)
    with open(os.path.join(tmp_path, "wandb-metadata.json"), "w") as f:
        json.dump(metadata, f)

    requirements = requirements or []
    with open(os.path.join(tmp_path, "requirements.txt"), "w") as f:
        f.write("\n".join(requirements))


def _clean_python_version(python_version: str) -> str:
    # remove micro if present
    if python_version.count(".") > 1:
        python_version = ".".join(python_version.split(".")[:2])
        _logger.debug(f"micro python version stripped. Now: {python_version}")
    return python_version