Run Module¶

This section documents the run components of the Nextmv Python SDK.

run ¶

This module contains definitions for an app run.

CLASS	DESCRIPTION
`Metadata`	Metadata of a run, whether it was successful or not.
`RunInformation`	Information of a run.
`ErrorLog`	Error log of a run, when it was not successful.
`RunResult`	Result of a run, whether it was successful or not.
`RunLog`	Log of a run.
`FormatInput`	Input format for a run configuration.
`FormatOutput`	Output format for a run configuration.
`Format`	Format for a run configuration.
`RunType`	The actual type of the run.
`RunTypeConfiguration`	Defines the configuration for the type of the run that is being executed on an application.
`RunQueuing`	RunQueuing configuration for a run.
`RunConfiguration`	Configuration for an app run.
`ExternalRunResult`	Result of a run used to configure a new application run as an external one.
`TrackedRunStatus`	The status of a tracked run.
`TrackedRun`	An external run that is tracked in the Nextmv platform.

FUNCTION	DESCRIPTION
`run_duration`	Calculate the duration of a run in milliseconds.

ErrorLog ¶

Bases: BaseModel

Error log of a run, when it was not successful.

You can import the ErrorLog class directly from nextmv:

from nextmv import ErrorLog

PARAMETER	DESCRIPTION
`error` ¶	Error message. Defaults to None. TYPE: `str`
`stdout` ¶	Standard output. Defaults to None. TYPE: `str`
`stderr` ¶	Standard error. Defaults to None. TYPE: `str`

error `class-attribute` `instance-attribute` ¶

error: Optional[str] = None

Error message.

stderr `class-attribute` `instance-attribute` ¶

stderr: Optional[str] = None

Standard error.

stdout `class-attribute` `instance-attribute` ¶

stdout: Optional[str] = None

Standard output.

ExternalRunResult ¶

Bases: BaseModel

Result of a run used to configure a new application run as an external one.

You can import the ExternalRunResult class directly from nextmv:

from nextmv import ExternalRunResult

PARAMETER	DESCRIPTION
`output_upload_id` ¶	ID of the output upload. Defaults to None. TYPE: `str`
`error_upload_id` ¶	ID of the error upload. Defaults to None. TYPE: `str`
`status` ¶	Status of the run. Must be "succeeded" or "failed". Defaults to None. TYPE: `str`
`error_message` ¶	Error message of the run. Defaults to None. TYPE: `str`
`execution_duration` ¶	Duration of the run, in milliseconds. Defaults to None. TYPE: `int`

Examples:

>>> from nextmv import ExternalRunResult
>>> # Successful external run
>>> result = ExternalRunResult(
...     output_upload_id="upload-12345",
...     status="succeeded",
...     execution_duration=5000
... )
>>> result.status
'succeeded'
>>> result.execution_duration
5000

>>> # Failed external run
>>> failed_result = ExternalRunResult(
...     error_upload_id="error-67890",
...     status="failed",
...     error_message="Optimization failed due to invalid constraints",
...     execution_duration=2000
... )
>>> failed_result.status
'failed'
>>> failed_result.error_message
'Optimization failed due to invalid constraints'

error_message `class-attribute` `instance-attribute` ¶

error_message: Optional[str] = None

Error message of the run.

error_upload_id `class-attribute` `instance-attribute` ¶

error_upload_id: Optional[str] = None

ID of the error upload.

execution_duration `class-attribute` `instance-attribute` ¶

execution_duration: Optional[int] = None

Duration of the run, in milliseconds.

output_upload_id `class-attribute` `instance-attribute` ¶

output_upload_id: Optional[str] = None

ID of the output upload.

status `class-attribute` `instance-attribute` ¶

status: Optional[str] = None

Status of the run.

Format ¶

Bases: BaseModel

Format for a run configuration.

You can import the Format class directly from nextmv:

from nextmv import Format

PARAMETER	DESCRIPTION
`format_input` ¶	Input format for the run configuration. TYPE: `FormatInput`
`format_output` ¶	Output format for the run configuration. Defaults to None. TYPE: `FormatOutput`

Examples:

>>> from nextmv import Format, FormatInput, FormatOutput, InputFormat, OutputFormat
>>> format_config = Format(
...     format_input=FormatInput(input_type=InputFormat.JSON),
...     format_output=FormatOutput(output_type=OutputFormat.JSON)
... )
>>> format_config.format_input.input_type
<InputFormat.JSON: 'json'>
>>> format_config.format_output.output_type
<OutputFormat.JSON: 'json'>

format_input `class-attribute` `instance-attribute` ¶

format_input: FormatInput = Field(
    serialization_alias="input",
    validation_alias=AliasChoices("input", "format_input"),
)

Input format for the run configuration.

format_output `class-attribute` `instance-attribute` ¶

format_output: Optional[FormatOutput] = Field(
    serialization_alias="output",
    validation_alias=AliasChoices(
        "output", "format_output"
    ),
    default=None,
)

Output format for the run configuration.

FormatInput ¶

Bases: BaseModel

Input format for a run configuration.

You can import the FormatInput class directly from nextmv:

from nextmv import FormatInput

PARAMETER	DESCRIPTION
`input_type` ¶	Type of the input format. Defaults to `InputFormat.JSON`. TYPE: `InputFormat`

Examples:

>>> from nextmv import FormatInput, InputFormat
>>> format_input = FormatInput()
>>> format_input.input_type
<InputFormat.JSON: 'json'>

>>> format_input = FormatInput(input_type=InputFormat.TEXT)
>>> format_input.input_type
<InputFormat.TEXT: 'text'>

input_type `class-attribute` `instance-attribute` ¶

input_type: InputFormat = Field(
    serialization_alias="type",
    validation_alias=AliasChoices("type", "input_type"),
    default=JSON,
)

Type of the input format.

FormatOutput ¶

Bases: BaseModel

Output format for a run configuration.

You can import the FormatOutput class directly from nextmv:

from nextmv import FormatOutput

PARAMETER	DESCRIPTION
`output_type` ¶	Type of the output format. Defaults to `OutputFormat.JSON`. TYPE: `OutputFormat`

Examples:

>>> from nextmv import FormatOutput, OutputFormat
>>> format_output = FormatOutput()
>>> format_output.output_type
<OutputFormat.JSON: 'json'>

>>> format_output = FormatOutput(output_type=OutputFormat.CSV_ARCHIVE)
>>> format_output.output_type
<OutputFormat.CSV_ARCHIVE: 'csv_archive'>

output_type `class-attribute` `instance-attribute` ¶

output_type: OutputFormat = Field(
    serialization_alias="type",
    validation_alias=AliasChoices("type", "output_type"),
    default=JSON,
)

Type of the output format.

Metadata ¶

Bases: BaseModel

Metadata of a run, whether it was successful or not.

You can import the Metadata class directly from nextmv:

from nextmv import Metadata

PARAMETER	DESCRIPTION
`application_id` ¶	ID of the application where the run was submitted to. TYPE: `str`
`application_instance_id` ¶	ID of the instance where the run was submitted to. TYPE: `str`
`application_version_id` ¶	ID of the version of the application where the run was submitted to. TYPE: `str`
`created_at` ¶	Date and time when the run was created. TYPE: `datetime`
`duration` ¶	Duration of the run in milliseconds. TYPE: `float`
`error` ¶	Error message if the run failed. TYPE: `str`
`input_size` ¶	Size of the input in bytes. TYPE: `float`
`output_size` ¶	Size of the output in bytes. TYPE: `float`
`format` ¶	Format of the input and output of the run. TYPE: `Format`
`status` ¶	Deprecated: use status_v2. TYPE: `Status`
`status_v2` ¶	Status of the run. TYPE: `StatusV2`

application_id `instance-attribute` ¶

application_id: str

ID of the application where the run was submitted to.

application_instance_id `instance-attribute` ¶

application_instance_id: str

ID of the instance where the run was submitted to.

application_version_id `instance-attribute` ¶

application_version_id: str

ID of the version of the application where the run was submitted to.

created_at `instance-attribute` ¶

created_at: datetime

Date and time when the run was created.

duration `instance-attribute` ¶

duration: float

Duration of the run in milliseconds.

error `instance-attribute` ¶

error: str

Error message if the run failed.

format `instance-attribute` ¶

format: Format

Format of the input and output of the run.

input_size `instance-attribute` ¶

input_size: float

Size of the input in bytes.

output_size `instance-attribute` ¶

output_size: float

Size of the output in bytes.

status `class-attribute` `instance-attribute` ¶

status: Optional[Status] = None

Deprecated: use status_v2.

status_v2 `instance-attribute` ¶

status_v2: StatusV2

Status of the run.

RunConfiguration ¶

Bases: BaseModel

Configuration for an app run.

You can import the RunConfiguration class directly from nextmv:

from nextmv import RunConfiguration

PARAMETER	DESCRIPTION
`execution_class` ¶	Execution class for the instance. Defaults to None. TYPE: `str`
`format` ¶	Format for the run configuration. Defaults to None. TYPE: `Format`
`run_type` ¶	Run type configuration for the run. Defaults to None. TYPE: `RunTypeConfiguration`
`secrets_collection_id` ¶	ID of the secrets collection to use for the run. Defaults to None. TYPE: `str`
`queuing` ¶	Queuing configuration for the run. Defaults to None. TYPE: `RunQueuing`

Examples:

>>> from nextmv import RunConfiguration, RunQueuing
>>> config = RunConfiguration(
...     execution_class="large",
...     queuing=RunQueuing(priority=1)
... )
>>> config.execution_class
'large'
>>> config.queuing.priority
1

>>> # Basic configuration
>>> basic_config = RunConfiguration()
>>> basic_config.format is None
True

execution_class `class-attribute` `instance-attribute` ¶

execution_class: Optional[str] = None

Execution class for the instance.

format `class-attribute` `instance-attribute` ¶

format: Optional[Format] = None

Format for the run configuration.

queuing `class-attribute` `instance-attribute` ¶

queuing: Optional[RunQueuing] = None

Queuing configuration for the run.

resolve ¶

resolve(
    input: Union[Input, dict[str, Any], BaseModel, str],
    dir_path: Optional[str] = None,
) -> None

Resolves the run configuration by modifying or setting the format, based on the type of input that is provided.

PARAMETER	DESCRIPTION
`input` ¶	The input to use for resolving the run configuration. TYPE: `Input or dict[str, Any] or BaseModel or str`
`dir_path` ¶	The directory path where inputs can be loaded from. TYPE: `str` DEFAULT: `None`

Examples:

>>> from nextmv import RunConfiguration
>>> config = RunConfiguration()
>>> config.resolve({"key": "value"})
>>> config.format.format_input.input_type
<InputFormat.JSON: 'json'>

>>> config = RunConfiguration()
>>> config.resolve("text input")
>>> config.format.format_input.input_type
<InputFormat.TEXT: 'text'>

>>> config = RunConfiguration()
>>> config.resolve({}, dir_path="/path/to/files")
>>> config.format.format_input.input_type
<InputFormat.MULTI_FILE: 'multi_file'>

Source code in nextmv/nextmv/run.py

def resolve(
    self,
    input: Union[Input, dict[str, Any], BaseModel, str],
    dir_path: Optional[str] = None,
) -> None:
    """
    Resolves the run configuration by modifying or setting the `format`,
    based on the type of input that is provided.

    Parameters
    ----------
    input : Input or dict[str, Any] or BaseModel or str
        The input to use for resolving the run configuration.
    dir_path : str, optional
        The directory path where inputs can be loaded from.

    Examples
    --------
    >>> from nextmv import RunConfiguration
    >>> config = RunConfiguration()
    >>> config.resolve({"key": "value"})
    >>> config.format.format_input.input_type
    <InputFormat.JSON: 'json'>

    >>> config = RunConfiguration()
    >>> config.resolve("text input")
    >>> config.format.format_input.input_type
    <InputFormat.TEXT: 'text'>

    >>> config = RunConfiguration()
    >>> config.resolve({}, dir_path="/path/to/files")
    >>> config.format.format_input.input_type
    <InputFormat.MULTI_FILE: 'multi_file'>
    """

    # If the value is set by the user, do not change it.
    if self.format is not None:
        return

    self.format = Format(
        format_input=FormatInput(input_type=InputFormat.JSON),
        format_output=FormatOutput(output_type=OutputFormat.JSON),
    )

    if isinstance(input, dict):
        self.format.format_input.input_type = InputFormat.JSON
    elif isinstance(input, str):
        self.format.format_input.input_type = InputFormat.TEXT
    elif dir_path is not None and dir_path != "":
        # Kinda hard to detect if we should be working with CSV_ARCHIVE or
        # MULTI_FILE, so we default to MULTI_FILE.
        self.format.format_input.input_type = InputFormat.MULTI_FILE
    elif isinstance(input, Input):
        self.format.format_input.input_type = input.input_format

    # As input and output are symmetric, we set the output according to the input
    # format.
    if self.format.format_input.input_type == InputFormat.JSON:
        self.format.format_output = FormatOutput(output_type=OutputFormat.JSON)
    elif self.format.format_input.input_type == InputFormat.TEXT:  # Text still maps to json
        self.format.format_output = FormatOutput(output_type=OutputFormat.JSON)
    elif self.format.format_input.input_type == InputFormat.CSV_ARCHIVE:
        self.format.format_output = FormatOutput(output_type=OutputFormat.CSV_ARCHIVE)
    elif self.format.format_input.input_type == InputFormat.MULTI_FILE:
        self.format.format_output = FormatOutput(output_type=OutputFormat.MULTI_FILE)
    else:
        self.format.format_output = FormatOutput(output_type=OutputFormat.JSON)

run_type `class-attribute` `instance-attribute` ¶

run_type: Optional[RunTypeConfiguration] = None

Run type configuration for the run.

secrets_collection_id `class-attribute` `instance-attribute` ¶

secrets_collection_id: Optional[str] = None

ID of the secrets collection to use for the run.

RunInformation ¶

Bases: BaseModel

Information of a run.

You can import the RunInformation class directly from nextmv:

from nextmv import RunInformation

PARAMETER	DESCRIPTION
`description` ¶	Description of the run. TYPE: `str`
`id` ¶	ID of the run. TYPE: `str`
`metadata` ¶	Metadata of the run. TYPE: `Metadata`
`name` ¶	Name of the run. TYPE: `str`
`user_email` ¶	Email of the user who submitted the run. TYPE: `str`
`console_url` ¶	URL to the run in the Nextmv console. Defaults to "". TYPE: `str`

console_url `class-attribute` `instance-attribute` ¶

console_url: str = Field(default='')

URL to the run in the Nextmv console.

description `instance-attribute` ¶

description: str

Description of the run.

id `instance-attribute` ¶

id: str

ID of the run.

metadata `instance-attribute` ¶

metadata: Metadata

Metadata of the run.

name `instance-attribute` ¶

name: str

Name of the run.

synced_at `class-attribute` `instance-attribute` ¶

synced_at: Optional[datetime] = None

Timestamp when the run was synced with the remote run. This field is None if the run was not created using Application.sync or if the run has not been synced yet.

synced_run_id `class-attribute` `instance-attribute` ¶

synced_run_id: Optional[str] = None

ID of the synced remote run, if applicable. When the Application.sync method is used, this field marks the association between the local run (id) and the remote run (synced_run_id). This field is None if the run was not created using Application.sync or if the run has not been synced yet.

user_email `instance-attribute` ¶

user_email: str

Email of the user who submitted the run.

RunLog ¶

Bases: BaseModel

Log of a run.

You can import the RunLog class directly from nextmv:

from nextmv import RunLog

PARAMETER	DESCRIPTION
`log` ¶	Log of the run. TYPE: `str`

Examples:

>>> from nextmv import RunLog
>>> run_log = RunLog(log="Optimization completed successfully")
>>> run_log.log
'Optimization completed successfully'

>>> # Multi-line log
>>> multi_line_log = RunLog(log="Starting optimization\nProcessing data\nCompleted")
>>> multi_line_log.log
'Starting optimization\nProcessing data\nCompleted'

log `instance-attribute` ¶

log: str

Log of the run.

RunQueuing ¶

Bases: BaseModel

RunQueuing configuration for a run.

You can import the RunQueuing class directly from nextmv:

from nextmv import RunQueuing

PARAMETER	DESCRIPTION
`priority` ¶	Priority of the run in the queue. 1 is the highest priority, 9 is the lowest priority. Defaults to None. TYPE: `int`
`disabled` ¶	Whether the run should be queued, or not. If True, the run will not be queued. If False, the run will be queued. Defaults to None. TYPE: `bool`

Examples:

>>> from nextmv import RunQueuing
>>> queuing = RunQueuing(priority=1, disabled=False)
>>> queuing.priority
1
>>> queuing.disabled
False

>>> # High priority run
>>> high_priority = RunQueuing(priority=1)
>>> high_priority.priority
1

>>> # Disabled queuing
>>> no_queue = RunQueuing(disabled=True)
>>> no_queue.disabled
True

disabled `class-attribute` `instance-attribute` ¶

disabled: Optional[bool] = None

Whether the run should be queued, or not. If True, the run will not be queued. If False, the run will be queued.

priority `class-attribute` `instance-attribute` ¶

priority: Optional[int] = None

Priority of the run in the queue. 1 is the highest priority, 9 is the lowest priority.

RunResult ¶

Bases: RunInformation

Result of a run, whether it was successful or not.

You can import the RunResult class directly from nextmv:

from nextmv import RunResult

PARAMETER	DESCRIPTION
`error_log` ¶	Error log of the run. Only available if the run failed. Defaults to None. TYPE: `ErrorLog`
`output` ¶	Output of the run. Only available if the run succeeded. Defaults to None. TYPE: `dict[str, Any]`

console_url `class-attribute` `instance-attribute` ¶

console_url: str = Field(default='')

URL to the run in the Nextmv console.

description `instance-attribute` ¶

description: str

Description of the run.

error_log `class-attribute` `instance-attribute` ¶

error_log: Optional[ErrorLog] = None

Error log of the run. Only available if the run failed.

id `instance-attribute` ¶

id: str

ID of the run.

metadata `instance-attribute` ¶

metadata: Metadata

Metadata of the run.

name `instance-attribute` ¶

name: str

Name of the run.

output `class-attribute` `instance-attribute` ¶

output: Optional[dict[str, Any]] = None

Output of the run. Only available if the run succeeded.

synced_at `class-attribute` `instance-attribute` ¶

synced_at: Optional[datetime] = None

Timestamp when the run was synced with the remote run. This field is None if the run was not created using Application.sync or if the run has not been synced yet.

synced_run_id `class-attribute` `instance-attribute` ¶

synced_run_id: Optional[str] = None

ID of the synced remote run, if applicable. When the Application.sync method is used, this field marks the association between the local run (id) and the remote run (synced_run_id). This field is None if the run was not created using Application.sync or if the run has not been synced yet.

user_email `instance-attribute` ¶

user_email: str

Email of the user who submitted the run.

RunType ¶

Bases: str, Enum

The actual type of the run.

You can import the RunType class directly from nextmv:

from nextmv import RunType

PARAMETER	DESCRIPTION
`STANDARD` ¶	Standard run type. TYPE: `str`
`EXTERNAL` ¶	External run type. TYPE: `str`
`ENSEMBLE` ¶	Ensemble run type. TYPE: `str`

Examples:

>>> from nextmv import RunType
>>> run_type = RunType.STANDARD
>>> run_type
<RunType.STANDARD: 'standard'>
>>> run_type.value
'standard'

>>> # Creating from string
>>> external_type = RunType("external")
>>> external_type
<RunType.EXTERNAL: 'external'>

>>> # All available types
>>> list(RunType)
[<RunType.STANDARD: 'standard'>, <RunType.EXTERNAL: 'external'>, <RunType.ENSEMBLE: 'ensemble'>]

ENSEMBLE `class-attribute` `instance-attribute` ¶

ENSEMBLE = 'ensemble'

Ensemble run type.

EXTERNAL `class-attribute` `instance-attribute` ¶

EXTERNAL = 'external'

External run type.

STANDARD `class-attribute` `instance-attribute` ¶

STANDARD = 'standard'

Standard run type.

RunTypeConfiguration ¶

Bases: BaseModel

Defines the configuration for the type of the run that is being executed on an application.

You can import the RunTypeConfiguration class directly from nextmv:

from nextmv import RunTypeConfiguration

PARAMETER	DESCRIPTION
`run_type` ¶	Type of the run. TYPE: `RunType`
`definition_id` ¶	ID of the definition for the run type. Defaults to None. TYPE: `str`
`reference_id` ¶	ID of the reference for the run type. Defaults to None. TYPE: `str`

Examples:

>>> from nextmv import RunTypeConfiguration, RunType
>>> config = RunTypeConfiguration(run_type=RunType.STANDARD)
>>> config.run_type
<RunType.STANDARD: 'standard'>
>>> config.definition_id is None
True

>>> # External run with reference
>>> external_config = RunTypeConfiguration(
...     run_type=RunType.EXTERNAL,
...     reference_id="ref-12345"
... )
>>> external_config.run_type
<RunType.EXTERNAL: 'external'>
>>> external_config.reference_id
'ref-12345'

>>> # Ensemble run with definition
>>> ensemble_config = RunTypeConfiguration(
...     run_type=RunType.ENSEMBLE,
...     definition_id="def-67890"
... )
>>> ensemble_config.run_type
<RunType.ENSEMBLE: 'ensemble'>
>>> ensemble_config.definition_id
'def-67890'

definition_id `class-attribute` `instance-attribute` ¶

definition_id: Optional[str] = None

ID of the definition for the run type.

reference_id `class-attribute` `instance-attribute` ¶

reference_id: Optional[str] = None

ID of the reference for the run type.

run_type `class-attribute` `instance-attribute` ¶

run_type: RunType = Field(
    serialization_alias="type",
    validation_alias=AliasChoices("type", "run_type"),
)

Type of the run.

TrackedRun `dataclass` ¶

TrackedRun(
    status: TrackedRunStatus,
    input: Optional[
        Union[Input, dict[str, Any], str]
    ] = None,
    output: Optional[
        Union[Output, dict[str, Any], str]
    ] = None,
    duration: Optional[int] = None,
    error: Optional[str] = None,
    logs: Optional[list[str]] = None,
    name: Optional[str] = None,
    description: Optional[str] = None,
    input_dir_path: Optional[str] = None,
    output_dir_path: Optional[str] = None,
)

An external run that is tracked in the Nextmv platform.

You can import the TrackedRun class directly from nextmv:

from nextmv import TrackedRun

PARAMETER	DESCRIPTION
`status` ¶	The status of the run being tracked. This field is required. TYPE: `TrackedRunStatus`
`input` ¶	The input of the run being tracked. Please note that if the input format is JSON, then the input data must be JSON serializable. If both `input` and `input_dir_path` are specified, the `input` is ignored, and the files in the directory are used instead. Defaults to None. TYPE: `Input or dict[str, Any] or str` DEFAULT: `None`
`output` ¶	The output of the run being tracked. Please note that if the output format is JSON, then the output data must be JSON serializable. If both `output` and `output_dir_path` are specified, the `output` is ignored, and the files in the directory are used instead. Defaults to None. TYPE: `Output or dict[str, Any] or str` DEFAULT: `None`
`duration` ¶	The duration of the run being tracked, in milliseconds. This field is optional. Defaults to None. TYPE: `int` DEFAULT: `None`
`error` ¶	An error message if the run failed. You should only specify this if the run failed (the `status` is `TrackedRunStatus.FAILED`), otherwise an exception will be raised. This field is optional. Defaults to None. TYPE: `str` DEFAULT: `None`
`logs` ¶	The logs of the run being tracked. Each element of the list is a line in the log. This field is optional. Defaults to None. TYPE: `list[str]` DEFAULT: `None`
`name` ¶	Optional name for the run being tracked. Defaults to None. TYPE: `str` DEFAULT: `None`
`description` ¶	Optional description for the run being tracked. Defaults to None. TYPE: `str` DEFAULT: `None`
`input_dir_path` ¶	Path to a directory containing input files. If specified, the calling function will package the files in the directory into a tar file and upload it as a large input. This is useful for non-JSON input formats, such as when working with `CSV_ARCHIVE` or `MULTI_FILE`. If both `input` and `input_dir_path` are specified, the `input` is ignored, and the files in the directory are used instead. Defaults to None. TYPE: `str` DEFAULT: `None`
`output_dir_path` ¶	Path to a directory containing output files. If specified, the calling function will package the files in the directory into a tar file and upload it as a large output. This is useful for non-JSON output formats, such as when working with `CSV_ARCHIVE` or `MULTI_FILE`. If both `output` and `output_dir_path` are specified, the `output` is ignored, and the files are saved in the directory instead. Defaults to None. TYPE: `str` DEFAULT: `None`

Examples:

>>> from nextmv import TrackedRun, TrackedRunStatus
>>> # Successful run
>>> run = TrackedRun(
...     status=TrackedRunStatus.SUCCEEDED,
...     input={"vehicles": 5, "locations": 10},
...     output={"routes": [{"stops": [1, 2, 3]}]},
...     duration=5000,
...     name="test-run",
...     description="A test optimization run"
... )
>>> run.status
<TrackedRunStatus.SUCCEEDED: 'succeeded'>
>>> run.duration
5000

>>> # Failed run with error
>>> failed_run = TrackedRun(
...     status=TrackedRunStatus.FAILED,
...     input={"vehicles": 0},
...     error="No vehicles available for routing",
...     duration=1000,
...     logs=["Starting optimization", "Error: No vehicles found"]
... )
>>> failed_run.status
<TrackedRunStatus.FAILED: 'failed'>
>>> failed_run.error
'No vehicles available for routing'

>>> # Run with directory-based input/output
>>> dir_run = TrackedRun(
...     status=TrackedRunStatus.SUCCEEDED,
...     input_dir_path="/path/to/input/files",
...     output_dir_path="/path/to/output/files",
...     duration=10000
... )
>>> dir_run.input_dir_path
'/path/to/input/files'

RAISES	DESCRIPTION
`ValueError`	If the status value is invalid, if an error message is provided for a successful run, or if input/output formats are not JSON or input/output dicts are not JSON serializable.

description `class-attribute` `instance-attribute` ¶

description: Optional[str] = None

Optional description for the run being tracked.

duration `class-attribute` `instance-attribute` ¶

duration: Optional[int] = None

The duration of the run being tracked, in milliseconds.

error `class-attribute` `instance-attribute` ¶

error: Optional[str] = None

An error message if the run failed. You should only specify this if the run failed, otherwise an exception will be raised.

input `class-attribute` `instance-attribute` ¶

input: Optional[Union[Input, dict[str, Any], str]] = None

The input of the run being tracked. Please note that if the input format is JSON, then the input data must be JSON serializable. If both input and input_dir_path are specified, the input is ignored, and the files in the directory are used instead.

input_dir_path `class-attribute` `instance-attribute` ¶

input_dir_path: Optional[str] = None

Path to a directory containing input files. If specified, the calling function will package the files in the directory into a tar file and upload it as a large input. This is useful for non-JSON input formats, such as when working with CSV_ARCHIVE or MULTI_FILE. If both input and input_dir_path are specified, the input is ignored, and the files in the directory are used instead.

logs `class-attribute` `instance-attribute` ¶

logs: Optional[list[str]] = None

The logs of the run being tracked. Each element of the list is a line in the log.

logs_text ¶

logs_text() -> str

Returns the logs as a single string.

Each log entry is separated by a newline character.

RETURNS	DESCRIPTION
`str`	The logs as a single string. If no logs are present, an empty string is returned.

Examples:

>>> from nextmv import TrackedRun, TrackedRunStatus
>>> run = TrackedRun(
...     status=TrackedRunStatus.SUCCEEDED,
...     logs=["Starting optimization", "Processing data", "Optimization complete"]
... )
>>> run.logs_text()
'Starting optimization\nProcessing data\nOptimization complete'

>>> # Single string log
>>> run_with_string_log = TrackedRun(
...     status=TrackedRunStatus.SUCCEEDED,
...     logs="Single log entry"
... )
>>> run_with_string_log.logs_text()
'Single log entry'

>>> # No logs
>>> run_no_logs = TrackedRun(status=TrackedRunStatus.SUCCEEDED)
>>> run_no_logs.logs_text()
''

RAISES	DESCRIPTION
`TypeError`	If `self.logs` is not a string or a list of strings.

Source code in nextmv/nextmv/run.py

def logs_text(self) -> str:
    """
    Returns the logs as a single string.

    Each log entry is separated by a newline character.

    Returns
    -------
    str
        The logs as a single string. If no logs are present, an empty
        string is returned.

    Examples
    --------
    >>> from nextmv import TrackedRun, TrackedRunStatus
    >>> run = TrackedRun(
    ...     status=TrackedRunStatus.SUCCEEDED,
    ...     logs=["Starting optimization", "Processing data", "Optimization complete"]
    ... )
    >>> run.logs_text()
    'Starting optimization\\nProcessing data\\nOptimization complete'

    >>> # Single string log
    >>> run_with_string_log = TrackedRun(
    ...     status=TrackedRunStatus.SUCCEEDED,
    ...     logs="Single log entry"
    ... )
    >>> run_with_string_log.logs_text()
    'Single log entry'

    >>> # No logs
    >>> run_no_logs = TrackedRun(status=TrackedRunStatus.SUCCEEDED)
    >>> run_no_logs.logs_text()
    ''

    Raises
    ------
    TypeError
        If `self.logs` is not a string or a list of strings.
    """

    if self.logs is None:
        return ""

    if isinstance(self.logs, str):
        return self.logs

    if isinstance(self.logs, list):
        return "\\n".join(self.logs)

    raise TypeError("Logs must be a string or a list of strings.")

name `class-attribute` `instance-attribute` ¶

name: Optional[str] = None

Optional name for the run being tracked.

output `class-attribute` `instance-attribute` ¶

output: Optional[Union[Output, dict[str, Any], str]] = None

The output of the run being tracked. Please note that if the output format is JSON, then the output data must be JSON serializable. If both output and output_dir_path are specified, the output is ignored, and the files in the directory are used instead.

output_dir_path `class-attribute` `instance-attribute` ¶

output_dir_path: Optional[str] = None

Path to a directory containing output files. If specified, the calling function will package the files in the directory into a tar file and upload it as a large output. This is useful for non-JSON output formats, such as when working with CSV_ARCHIVE or MULTI_FILE. If both output and output_dir_path are specified, the output is ignored, and the files are saved in the directory instead.

status `instance-attribute` ¶

status: TrackedRunStatus

The status of the run being tracked

TrackedRunStatus ¶

Bases: str, Enum

The status of a tracked run.

You can import the TrackedRunStatus class directly from nextmv:

from nextmv import TrackedRunStatus

PARAMETER	DESCRIPTION
`SUCCEEDED` ¶	The run succeeded. TYPE: `str`
`FAILED` ¶	The run failed. TYPE: `str`

Examples:

>>> from nextmv import TrackedRunStatus
>>> status = TrackedRunStatus.SUCCEEDED
>>> status
<TrackedRunStatus.SUCCEEDED: 'succeeded'>
>>> status.value
'succeeded'

>>> # Creating from string
>>> failed_status = TrackedRunStatus("failed")
>>> failed_status
<TrackedRunStatus.FAILED: 'failed'>

>>> # All available statuses
>>> list(TrackedRunStatus)
[<TrackedRunStatus.SUCCEEDED: 'succeeded'>, <TrackedRunStatus.FAILED: 'failed'>]

FAILED `class-attribute` `instance-attribute` ¶

FAILED = 'failed'

The run failed.

SUCCEEDED `class-attribute` `instance-attribute` ¶

SUCCEEDED = 'succeeded'

The run succeeded.

run_duration ¶

run_duration(
    start: Union[datetime, float],
    end: Union[datetime, float],
) -> int

Calculate the duration of a run in milliseconds.

You can import the run_duration function directly from nextmv:

from nextmv import run_duration

PARAMETER	DESCRIPTION
`start` ¶	The start time of the run. Can be a datetime object or a float representing the start time in seconds since the epoch. TYPE: `datetime or float`
`end` ¶	The end time of the run. Can be a datetime object or a float representing the end time in seconds since the epoch. TYPE: `datetime or float`

RETURNS	DESCRIPTION
`int`	The duration of the run in milliseconds.

RAISES	DESCRIPTION
`ValueError`	If the start time is after the end time.
`TypeError`	If start and end are not both datetime objects or both float numbers.

Examples:

>>> from datetime import datetime, timedelta
>>> start_dt = datetime(2023, 1, 1, 12, 0, 0)
>>> end_dt = datetime(2023, 1, 1, 12, 0, 1)
>>> run_duration(start_dt, end_dt)
1000

>>> start_float = 1672574400.0  # Corresponds to 2023-01-01 12:00:00
>>> end_float = 1672574401.0    # Corresponds to 2023-01-01 12:00:01
>>> run_duration(start_float, end_float)
1000

Source code in nextmv/nextmv/run.py

def run_duration(start: Union[datetime, float], end: Union[datetime, float]) -> int:
    """
    Calculate the duration of a run in milliseconds.

    You can import the `run_duration` function directly from `nextmv`:

    ```python
    from nextmv import run_duration
    ```

    Parameters
    ----------
    start : datetime or float
        The start time of the run. Can be a datetime object or a float
        representing the start time in seconds since the epoch.
    end : datetime or float
        The end time of the run. Can be a datetime object or a float
        representing the end time in seconds since the epoch.

    Returns
    -------
    int
        The duration of the run in milliseconds.

    Raises
    ------
    ValueError
        If the start time is after the end time.
    TypeError
        If start and end are not both datetime objects or both float numbers.

    Examples
    --------
    >>> from datetime import datetime, timedelta
    >>> start_dt = datetime(2023, 1, 1, 12, 0, 0)
    >>> end_dt = datetime(2023, 1, 1, 12, 0, 1)
    >>> run_duration(start_dt, end_dt)
    1000

    >>> start_float = 1672574400.0  # Corresponds to 2023-01-01 12:00:00
    >>> end_float = 1672574401.0    # Corresponds to 2023-01-01 12:00:01
    >>> run_duration(start_float, end_float)
    1000
    """
    if isinstance(start, float) and isinstance(end, float):
        if start > end:
            raise ValueError("Start time must be before end time.")
        return int(round((end - start) * 1000))

    if isinstance(start, datetime) and isinstance(end, datetime):
        if start > end:
            raise ValueError("Start time must be before end time.")
        return int(round((end - start).total_seconds() * 1000))

    raise TypeError("Start and end must be either datetime or float.")

Run Module¶

run ¶

ErrorLog ¶

error ¶

stdout ¶

stderr ¶

error class-attribute instance-attribute ¶

stderr class-attribute instance-attribute ¶

stdout class-attribute instance-attribute ¶

ExternalRunResult ¶

output_upload_id ¶

error_upload_id ¶

status ¶

error_message ¶

execution_duration ¶

error_message class-attribute instance-attribute ¶

error_upload_id class-attribute instance-attribute ¶

execution_duration class-attribute instance-attribute ¶

output_upload_id class-attribute instance-attribute ¶

status class-attribute instance-attribute ¶

Format ¶

format_input ¶

format_output ¶

format_input class-attribute instance-attribute ¶

format_output class-attribute instance-attribute ¶

FormatInput ¶

input_type ¶

input_type class-attribute instance-attribute ¶

FormatOutput ¶

output_type ¶

output_type class-attribute instance-attribute ¶

Metadata ¶

application_id ¶

application_instance_id ¶

application_version_id ¶

created_at ¶

duration ¶

error ¶

input_size ¶

output_size ¶

format ¶

status ¶

status_v2 ¶

application_id instance-attribute ¶

application_instance_id instance-attribute ¶

application_version_id instance-attribute ¶

created_at instance-attribute ¶

duration instance-attribute ¶

error instance-attribute ¶

format instance-attribute ¶

input_size instance-attribute ¶

output_size instance-attribute ¶

status class-attribute instance-attribute ¶

status_v2 instance-attribute ¶

RunConfiguration ¶

execution_class ¶

format ¶

run_type ¶

secrets_collection_id ¶

queuing ¶

execution_class class-attribute instance-attribute ¶

format class-attribute instance-attribute ¶

queuing class-attribute instance-attribute ¶

resolve ¶

input ¶

dir_path ¶

run_type class-attribute instance-attribute ¶

secrets_collection_id class-attribute instance-attribute ¶

RunInformation ¶

description ¶

id ¶

metadata ¶

name ¶

user_email ¶

console_url ¶

console_url class-attribute instance-attribute ¶

description instance-attribute ¶

id instance-attribute ¶

metadata instance-attribute ¶

name instance-attribute ¶

`error` ¶

`stdout` ¶

`stderr` ¶

error `class-attribute` `instance-attribute` ¶

stderr `class-attribute` `instance-attribute` ¶

stdout `class-attribute` `instance-attribute` ¶

`output_upload_id` ¶

`error_upload_id` ¶

`status` ¶

`error_message` ¶

`execution_duration` ¶

error_message `class-attribute` `instance-attribute` ¶

error_upload_id `class-attribute` `instance-attribute` ¶

execution_duration `class-attribute` `instance-attribute` ¶

output_upload_id `class-attribute` `instance-attribute` ¶

status `class-attribute` `instance-attribute` ¶

`format_input` ¶

`format_output` ¶

format_input `class-attribute` `instance-attribute` ¶

format_output `class-attribute` `instance-attribute` ¶

`input_type` ¶

input_type `class-attribute` `instance-attribute` ¶

`output_type` ¶

output_type `class-attribute` `instance-attribute` ¶

`application_id` ¶

`application_instance_id` ¶

`application_version_id` ¶

`created_at` ¶

`duration` ¶

`error` ¶

`input_size` ¶

`output_size` ¶

`format` ¶

`status` ¶

`status_v2` ¶

application_id `instance-attribute` ¶

application_instance_id `instance-attribute` ¶

application_version_id `instance-attribute` ¶

created_at `instance-attribute` ¶

duration `instance-attribute` ¶

error `instance-attribute` ¶

format `instance-attribute` ¶

input_size `instance-attribute` ¶

output_size `instance-attribute` ¶

status `class-attribute` `instance-attribute` ¶

status_v2 `instance-attribute` ¶

`execution_class` ¶

`format` ¶

`run_type` ¶

`secrets_collection_id` ¶

`queuing` ¶

execution_class `class-attribute` `instance-attribute` ¶

format `class-attribute` `instance-attribute` ¶

queuing `class-attribute` `instance-attribute` ¶

`input` ¶

`dir_path` ¶

run_type `class-attribute` `instance-attribute` ¶

secrets_collection_id `class-attribute` `instance-attribute` ¶

`description` ¶

`id` ¶

`metadata` ¶

`name` ¶

`user_email` ¶

`console_url` ¶

console_url `class-attribute` `instance-attribute` ¶

description `instance-attribute` ¶

id `instance-attribute` ¶

metadata `instance-attribute` ¶

name `instance-attribute` ¶

synced_at `class-attribute` `instance-attribute` ¶

synced_run_id `class-attribute` `instance-attribute` ¶

user_email `instance-attribute` ¶

`log` ¶

log `instance-attribute` ¶

`priority` ¶

`disabled` ¶

disabled `class-attribute` `instance-attribute` ¶

priority `class-attribute` `instance-attribute` ¶

`error_log` ¶

`output` ¶