Runner module¶

def run(
    app_id: str,
    src: str,
    manifest: Manifest,
    run_config: dict[str, Any],
    name: str | None = None,
    description: str | None = None,
    input_data: dict[str, Any] | str | None = None,
    inputs_dir_path: str | None = None,
    options: dict[str, Any] | None = None,
    cloned_run_id: str | None = None,
) -> str:
    """
    Execute a local run.

    This method recreates, partially, what the Nextmv Cloud does in the backend
    when running an application. A run ID is generated, a run directory is
    created, and the input data is recorded. Then, a subprocess is started to
    execute the application run in a detached manner. This means that the
    application run is not waited upon.

    Parameters
    ----------
    app_id : str
        The ID of the application.
    src : str
        The path to the application source code.
    manifest : Manifest
        The application manifest.
    run_config : dict[str, Any]
        The run configuration.
    name : Optional[str], optional
        The name for the run, by default None.
    description : Optional[str], optional
        The description for the run, by default None.
    input_data : Optional[Union[dict[str, Any], str]], optional
        The input data for the run, by default None. If `inputs_dir_path` is
        provided, this parameter is ignored.
    inputs_dir_path : Optional[str], optional
        The path to the directory containing input files, by default None. If
        provided, this parameter takes precedence over `input_data`.
    options : Optional[dict[str, Any]], optional
        Additional options for the run, by default None.
    cloned_run_id : Optional[str], optional
        The ID of the run that is being cloned, if applicable, by default None.

    Returns
    -------
    str
        The ID of the created run.
    """

    # Initialize the run: create the ID, dir, and write the input.
    run_id = safe_id("local")
    run_dir = new_run(
        app_id=app_id,
        src=src,
        run_id=run_id,
        run_config=run_config,
        name=name,
        description=description,
        options=options,
        cloned_run_id=cloned_run_id,
    )
    record_input(
        run_dir=run_dir,
        run_id=run_id,
        input_data=input_data,
        inputs_dir_path=inputs_dir_path,
    )

    # Prepare manifest dictionary for execution. We need to convert relative paths to
    # absolute paths to allow execution in a different directory.
    manifest_dict = manifest.to_dict()
    pip_requirements = manifest_dict.get("python", {}).get("pip-requirements", None)
    if pip_requirements is not None and isinstance(pip_requirements, str) and pip_requirements.strip():
        manifest_dict["python"]["pip-requirements"] = os.path.abspath(os.path.join(src, pip_requirements))

    # Start the process as a daemon (detached) so we don't wait for it to
    # finish. We send the input via stdin and close it immediately without
    # waiting. We call the `executor.py` script to do the actual execution.
    stdin_input = json.dumps(
        {
            "run_id": run_id,
            "src": os.path.abspath(src),
            "manifest_dict": manifest_dict,
            "run_dir": os.path.abspath(run_dir),
            "run_config": run_config,
            "input_data": input_data,
            "inputs_dir_path": os.path.abspath(inputs_dir_path) if inputs_dir_path is not None else None,
            "options": options,
        }
    )
    # When frozen as a single binary (PyInstaller), sys.executable points to
    # the binary itself rather than a Python interpreter, so we use the
    # --run-script mechanism to run executor.py with the bundled interpreter.
    # In a normal Python package install, sys.executable is a real interpreter
    # and executor.py can be launched directly.
    # We always use the absolute path to executor.py so it can be resolved
    # regardless of the working directory the binary was invoked from.
    executor_path = os.path.join(os.path.dirname(__file__), "executor.py")
    if getattr(sys, "frozen", False):
        args = [sys.executable, "--run-script", executor_path]
    else:
        args = [sys.executable, executor_path]
    process = subprocess.Popen(
        args,
        env=os.environ,
        text=True,
        stdin=subprocess.PIPE,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        cwd=os.path.dirname(executor_path),
        start_new_session=True,
    )
    process.stdin.write(stdin_input)
    process.stdin.close()

    return run_id

def new_run(
    app_id: str,
    src: str,
    run_id: str,
    run_config: dict[str, Any],
    name: str | None = None,
    description: str | None = None,
    options: dict[str, Any] | None = None,
    cloned_run_id: str | None = None,
) -> str:
    """
    Initializes a new run.

    The run information is recorded in a JSON file within the run directory.

    Parameters
    ----------
    app_id : str
        The ID of the application.
    src : str
        The path to the application source code.
    run_id : str
        The ID of the run.
    run_config : dict[str, Any]
        The run configuration.
    name : Optional[str], optional
        The name for the run, by default None.
    description : Optional[str], optional
        The description for the run, by default None.
    options : Optional[dict[str, Any]], optional
        Additional options for the run, by default None.
    cloned_run_id : Optional[str], optional
        The ID of the run that is being cloned, if applicable, by default None.

    Returns
    -------
    str
        The path to the new run directory.
    """

    # First, ensure the runs directory exists.
    runs_dir = os.path.join(src, NEXTMV_DIR, RUNS_KEY)
    os.makedirs(runs_dir, exist_ok=True)
    nextmv_gitignore_path = os.path.join(src, NEXTMV_DIR, ".gitignore")
    if not os.path.exists(nextmv_gitignore_path):
        with open(nextmv_gitignore_path, "w") as f:
            f.write("*\n")
            f.write("!.gitignore\n")

    # Create a new run directory.
    run_dir = os.path.join(runs_dir, run_id)
    os.makedirs(run_dir, exist_ok=True)

    # Create the run information file.
    created_at = datetime.now(timezone.utc)
    req_opt = options if options is not None else {}
    metadata = Metadata(
        application_id=app_id,
        application_instance_id="",
        application_version_id="",
        created_at=created_at,
        duration=0.0,
        error="",
        execution_class="local",
        execution_duration=0.0,
        experiment_id="",
        experiment_type="",
        format=Format(
            format_input=FormatInput(
                input_type=run_config["format"]["input"]["type"],
            ),
            format_output=FormatOutput(
                output_type=run_config["format"]["input"]["type"],
            ),
        ),
        initiated_at=created_at,
        input_size=0.0,
        integration=None,
        metrics=None,
        options=RunOptions(
            active_options=req_opt,
            options_summary=[OptionSummary(name=k, source="run", value=v) for k, v in req_opt.items()],
            request_options=req_opt,
        ),
        output_size=0.0,
        queuing_disabled=True,
        queuing_priority=0,
        run_type=RunTypeConfiguration(
            definition_id="",
            reference_id="",
            run_type=RunType.STANDARD,
        ),
        runtime="local",
        secrets_collection_id="",
        status_v2=StatusV2.queued,
        tracking=RunTrackingMetadata(
            cloned_run_id=cloned_run_id if cloned_run_id else "",
            input_id="",
        ),
    )

    if description is None:
        description = f"Local run created at {created_at.isoformat().replace('+00:00', 'Z')}"

    if name is None or name == "":
        name = f"local run {run_id}"

    information = RunInformation(
        description=description,
        id=run_id,
        metadata=metadata,
        name=name,
        user_email="",
    )
    with open(os.path.join(run_dir, f"{run_id}.json"), "w") as f:
        json.dump(information.to_dict(), f, indent=2)

    return run_dir

def record_input(
    run_dir: str,
    run_id: str,
    input_data: dict[str, Any] | str | None = None,
    inputs_dir_path: str | None = None,
) -> None:
    """
    Writes the input to the appropriate location.

    The size of the input is calculated and recorded in the run information.

    Parameters
    ----------
    run_dir : str
        The path to the run directory.
    run_id : str
        The ID of the run.
    input_data : Optional[Union[dict[str, Any], str]], optional
        The input data for the run, by default None. If `inputs_dir_path` is
        provided, this parameter is ignored.
    inputs_dir_path : Optional[str], optional
        The path to the directory containing input files, by default None. If
        provided, this parameter takes precedence over `input_data`.
    """

    # Create the inputs directory.
    run_inputs_dir = os.path.join(run_dir, INPUTS_KEY)
    os.makedirs(run_inputs_dir, exist_ok=True)

    if inputs_dir_path is not None and inputs_dir_path != "":
        # If we specify an inputs directory, we ignore the input_data.
        # Copy all files from inputs_dir_path to run_inputs_dir
        if os.path.exists(inputs_dir_path) and os.path.isdir(inputs_dir_path):
            shutil.copytree(inputs_dir_path, run_inputs_dir, dirs_exist_ok=True)

    elif isinstance(input_data, dict):
        # If no inputs_dir_path is provided, try a single JSON input.
        with open(os.path.join(run_inputs_dir, DEFAULT_INPUT_JSON_FILE), "w") as f:
            json.dump(input_data, f, indent=2)

    elif isinstance(input_data, str):
        # If no inputs_dir_path is provided, try a single TEXT input.
        with open(os.path.join(run_inputs_dir, DEFAULT_INPUT_TEXT_FILE), "w") as f:
            f.write(input_data)

    else:
        raise ValueError(
            "Invalid input data type: input_data must be a dict or str, or inputs_dir_path must be provided."
        )

    # Update the input size in the run information file.
    calculate_files_size(run_dir, run_id, run_inputs_dir, metadata_key="input_size")