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.

OptionsSummaryItem

Bases: BaseModel

Summary item for options used in a run.

You can import the OptionsSummaryItem class directly from nextmv:

from nextmv import OptionsSummaryItem

PARAMETER	DESCRIPTION
name	Name of the option. TYPE: str
value	Value of the option. TYPE: Any
source	Source of the option. TYPE: str

Examples:

>>> from nextmv import OptionsSummaryItem
>>> option = OptionsSummaryItem(
...     name="time_limit",
...     value=30,
...     source="config"
... )
>>> option.name
'time_limit'
>>> option.value
30
>>> option.source
'config'

>>> # Option from environment variable
>>> env_option = OptionsSummaryItem(
...     name="solver_type",
...     value="gurobi",
...     source="environment"
... )
>>> env_option.source
'environment'

name instance-attribute

name: str

Name of the option.

source instance-attribute

source: str

Source of the option.

value instance-attribute

value: Any

Value of the option.

Run

Bases: BaseModel

Information about a run in the Nextmv platform.

You can import the Run class directly from nextmv:

from nextmv import Run

PARAMETER	DESCRIPTION
id	ID of the run. TYPE: str
user_email	Email of the user who initiated the run. TYPE: str
name	Name of the run. TYPE: str
description	Description of the run. TYPE: str
created_at	Timestamp when the run was created. TYPE: datetime
application_id	ID of the application associated with the run. TYPE: str
application_instance_id	ID of the application instance associated with the run. TYPE: str
application_version_id	ID of the application version associated with the run. TYPE: str
run_type	Configuration for the type of the run. TYPE: RunTypeConfiguration
execution_class	Class name for the execution of a job. TYPE: str
runtime	Runtime environment for the run. TYPE: str
status	Deprecated, use status_v2 instead. TYPE: Status
status_v2	Status of the run. TYPE: StatusV2
queuing_priority	Priority of the run in the queue. Defaults to None. TYPE: int
queuing_disabled	Whether the run is disabled from queuing. Defaults to None. TYPE: bool
experiment_id	ID of the experiment associated with the run. Defaults to None. TYPE: str
statistics	Statistics of the run. Defaults to None. TYPE: RunInfoStatistics
input_id	ID of the input associated with the run. Defaults to None. TYPE: str
option_set	ID of the option set associated with the run. Defaults to None. TYPE: str
options	Options associated with the run. Defaults to None. TYPE: dict[str, str]
request_options	Request options associated with the run. Defaults to None. TYPE: dict[str, str]
options_summary	Summary of options used in the run. Defaults to None. TYPE: list[OptionsSummaryItem]
scenario_id	ID of the scenario associated with the run. Defaults to None. TYPE: str
repetition	Repetition number of the run. Defaults to None. TYPE: int
input_set_id	ID of the input set associated with the run. Defaults to None. TYPE: str

Examples:

>>> from nextmv import Run, RunTypeConfiguration, RunType, StatusV2
>>> from datetime import datetime
>>> run = Run(
...     id="run-12345",
...     user_email="[email protected]",
...     name="Test Run",
...     description="A test optimization run",
...     created_at=datetime.now(),
...     application_id="app-123",
...     application_instance_id="instance-456",
...     application_version_id="version-789",
...     run_type=RunTypeConfiguration(run_type=RunType.STANDARD),
...     execution_class="small",
...     runtime="python",
...     status_v2=StatusV2.SUCCEEDED
... )
>>> run.id
'run-12345'
>>> run.name
'Test Run'

application_id instance-attribute

application_id: str

ID of the application associated with the run.

application_instance_id instance-attribute

application_instance_id: str

ID of the application instance associated with the run.

application_version_id instance-attribute

application_version_id: str

ID of the application version associated with the run.

created_at instance-attribute

created_at: datetime

Timestamp when the run was created.

description instance-attribute

description: str

Description of the run.

execution_class instance-attribute

execution_class: str

Class name for the execution of a job.

experiment_id class-attribute instance-attribute

experiment_id: Optional[str] = None

ID of the experiment associated with the run.

id instance-attribute

id: str

ID of the run.

input_id class-attribute instance-attribute

input_id: Optional[str] = None

ID of the input associated with the run.

input_set_id class-attribute instance-attribute

input_set_id: Optional[str] = None

ID of the input set associated with the run.

name instance-attribute

name: str

Name of the run.

option_set class-attribute instance-attribute

option_set: Optional[str] = None

ID of the option set associated with the run.

options class-attribute instance-attribute

options: Optional[dict[str, str]] = None

Options associated with the run.

options_summary class-attribute instance-attribute

options_summary: Optional[list[OptionsSummaryItem]] = None

Summary of options used in the run.

queuing_disabled class-attribute instance-attribute

queuing_disabled: Optional[bool] = None

Whether the run is disabled from queuing.

queuing_priority class-attribute instance-attribute

queuing_priority: Optional[int] = None

Priority of the run in the queue.

repetition class-attribute instance-attribute

repetition: Optional[int] = None

Repetition number of the run.

request_options class-attribute instance-attribute

request_options: Optional[dict[str, str]] = None

Request options associated with the run.

run_type instance-attribute

run_type: RunTypeConfiguration

Configuration for the type of the run.

runtime instance-attribute

runtime: str

Runtime environment for the run.

scenario_id class-attribute instance-attribute

scenario_id: Optional[str] = None

ID of the scenario associated with the run.

statistics class-attribute instance-attribute

statistics: Optional[RunInfoStatistics] = None

Statistics of the run.

status class-attribute instance-attribute

status: Optional[Status] = None

Deprecated, use status_v2 instead.

status_v2 instance-attribute

status_v2: StatusV2

Status of the run.

user_email instance-attribute

user_email: str

Email of the user who initiated 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.

RunInfoStatistics

Bases: BaseModel

Statistics information for a run.

You can import the RunInfoStatistics class directly from nextmv:

from nextmv import RunInfoStatistics

PARAMETER	DESCRIPTION
status	Status of the statistics in the run. TYPE: str
error	Error message if the statistics could not be retrieved. Defaults to None. TYPE: str
indicators	List of statistics indicators. Defaults to None. TYPE: list[StatisticsIndicator]

Examples:

>>> from nextmv import RunInfoStatistics, StatisticsIndicator
>>> indicators = [
...     StatisticsIndicator(name="total_cost", value=1250.75),
...     StatisticsIndicator(name="optimal", value=True)
... ]
>>> stats = RunInfoStatistics(status="success", indicators=indicators)
>>> stats.status
'success'
>>> len(stats.indicators)
2

>>> # Statistics with error
>>> error_stats = RunInfoStatistics(
...     status="error",
...     error="Failed to calculate statistics"
... )
>>> error_stats.status
'error'
>>> error_stats.error
'Failed to calculate statistics'

error class-attribute instance-attribute

error: Optional[str] = None

Error message if the statistics could not be retrieved.

indicators class-attribute instance-attribute

indicators: Optional[list[StatisticsIndicator]] = None

List of statistics indicators.

status instance-attribute

status: str

Status of the statistics in 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.

to_run

to_run() -> Run

Transform this RunInformation instance into a Run instance.

This method maps all available attributes from the RunInformation and its metadata to create a Run instance. Attributes that are not available in RunInformation are set to None or appropriate defaults.

RETURNS	DESCRIPTION
Run	A Run instance with attributes populated from this RunInformation.

Examples:

>>> from nextmv import RunInformation, Metadata, Format, FormatInput, FormatOutput
>>> from nextmv import StatusV2, RunTypeConfiguration, RunType
>>> from datetime import datetime
>>> metadata = Metadata(
...     application_id="app-123",
...     application_instance_id="instance-456",
...     application_version_id="version-789",
...     created_at=datetime.now(),
...     duration=5000.0,
...     error="",
...     input_size=1024.0,
...     output_size=2048.0,
...     format=Format(
...         format_input=FormatInput(),
...         format_output=FormatOutput()
...     ),
...     status_v2=StatusV2.SUCCEEDED
... )
>>> run_info = RunInformation(
...     id="run-123",
...     description="Test run",
...     name="Test",
...     user_email="[email protected]",
...     metadata=metadata
... )
>>> run = run_info.to_run()
>>> run.id
'run-123'
>>> run.application_id
'app-123'

Source code in nextmv/nextmv/run.py

def to_run(self) -> Run:
    """
    Transform this `RunInformation` instance into a `Run` instance.

    This method maps all available attributes from the `RunInformation`
    and its metadata to create a `Run` instance. Attributes that are not
    available in RunInformation are set to None or appropriate defaults.

    Returns
    -------
    Run
        A Run instance with attributes populated from this RunInformation.

    Examples
    --------
    >>> from nextmv import RunInformation, Metadata, Format, FormatInput, FormatOutput
    >>> from nextmv import StatusV2, RunTypeConfiguration, RunType
    >>> from datetime import datetime
    >>> metadata = Metadata(
    ...     application_id="app-123",
    ...     application_instance_id="instance-456",
    ...     application_version_id="version-789",
    ...     created_at=datetime.now(),
    ...     duration=5000.0,
    ...     error="",
    ...     input_size=1024.0,
    ...     output_size=2048.0,
    ...     format=Format(
    ...         format_input=FormatInput(),
    ...         format_output=FormatOutput()
    ...     ),
    ...     status_v2=StatusV2.SUCCEEDED
    ... )
    >>> run_info = RunInformation(
    ...     id="run-123",
    ...     description="Test run",
    ...     name="Test",
    ...     user_email="[email protected]",
    ...     metadata=metadata
    ... )
    >>> run = run_info.to_run()
    >>> run.id
    'run-123'
    >>> run.application_id
    'app-123'
    """
    return Run(
        id=self.id,
        user_email=self.user_email,
        name=self.name,
        description=self.description,
        created_at=self.metadata.created_at,
        application_id=self.metadata.application_id,
        application_instance_id=self.metadata.application_instance_id,
        application_version_id=self.metadata.application_version_id,
        run_type=RunTypeConfiguration(),  # Default empty configuration
        execution_class="",  # Not available in RunInformation
        runtime="",  # Not available in RunInformation
        status=self.metadata.status,
        status_v2=self.metadata.status_v2,
        # Optional fields that are not available in RunInformation
        queuing_priority=None,
        queuing_disabled=None,
        experiment_id=None,
        statistics=None,
        input_id=None,
        option_set=None,
        options=None,
        request_options=None,
        options_summary=None,
        scenario_id=None,
        repetition=None,
        input_set_id=None,
    )

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.

to_run

to_run() -> Run

Transform this RunInformation instance into a Run instance.

This method maps all available attributes from the RunInformation and its metadata to create a Run instance. Attributes that are not available in RunInformation are set to None or appropriate defaults.

RETURNS	DESCRIPTION
Run	A Run instance with attributes populated from this RunInformation.

Examples:

>>> from nextmv import RunInformation, Metadata, Format, FormatInput, FormatOutput
>>> from nextmv import StatusV2, RunTypeConfiguration, RunType
>>> from datetime import datetime
>>> metadata = Metadata(
...     application_id="app-123",
...     application_instance_id="instance-456",
...     application_version_id="version-789",
...     created_at=datetime.now(),
...     duration=5000.0,
...     error="",
...     input_size=1024.0,
...     output_size=2048.0,
...     format=Format(
...         format_input=FormatInput(),
...         format_output=FormatOutput()
...     ),
...     status_v2=StatusV2.SUCCEEDED
... )
>>> run_info = RunInformation(
...     id="run-123",
...     description="Test run",
...     name="Test",
...     user_email="[email protected]",
...     metadata=metadata
... )
>>> run = run_info.to_run()
>>> run.id
'run-123'
>>> run.application_id
'app-123'

Source code in nextmv/nextmv/run.py

def to_run(self) -> Run:
    """
    Transform this `RunInformation` instance into a `Run` instance.

    This method maps all available attributes from the `RunInformation`
    and its metadata to create a `Run` instance. Attributes that are not
    available in RunInformation are set to None or appropriate defaults.

    Returns
    -------
    Run
        A Run instance with attributes populated from this RunInformation.

    Examples
    --------
    >>> from nextmv import RunInformation, Metadata, Format, FormatInput, FormatOutput
    >>> from nextmv import StatusV2, RunTypeConfiguration, RunType
    >>> from datetime import datetime
    >>> metadata = Metadata(
    ...     application_id="app-123",
    ...     application_instance_id="instance-456",
    ...     application_version_id="version-789",
    ...     created_at=datetime.now(),
    ...     duration=5000.0,
    ...     error="",
    ...     input_size=1024.0,
    ...     output_size=2048.0,
    ...     format=Format(
    ...         format_input=FormatInput(),
    ...         format_output=FormatOutput()
    ...     ),
    ...     status_v2=StatusV2.SUCCEEDED
    ... )
    >>> run_info = RunInformation(
    ...     id="run-123",
    ...     description="Test run",
    ...     name="Test",
    ...     user_email="[email protected]",
    ...     metadata=metadata
    ... )
    >>> run = run_info.to_run()
    >>> run.id
    'run-123'
    >>> run.application_id
    'app-123'
    """
    return Run(
        id=self.id,
        user_email=self.user_email,
        name=self.name,
        description=self.description,
        created_at=self.metadata.created_at,
        application_id=self.metadata.application_id,
        application_instance_id=self.metadata.application_instance_id,
        application_version_id=self.metadata.application_version_id,
        run_type=RunTypeConfiguration(),  # Default empty configuration
        execution_class="",  # Not available in RunInformation
        runtime="",  # Not available in RunInformation
        status=self.metadata.status,
        status_v2=self.metadata.status_v2,
        # Optional fields that are not available in RunInformation
        queuing_priority=None,
        queuing_disabled=None,
        experiment_id=None,
        statistics=None,
        input_id=None,
        option_set=None,
        options=None,
        request_options=None,
        options_summary=None,
        scenario_id=None,
        repetition=None,
        input_set_id=None,
    )

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: Optional[RunType] = Field(
    serialization_alias="type",
    validation_alias=AliasChoices("type", "run_type"),
    default=None,
)

Type of the run.

validate_run_type classmethod

validate_run_type(v)

Convert empty string to None for run_type validation.

Source code in nextmv/nextmv/run.py

@field_validator("run_type", mode="before")
@classmethod
def validate_run_type(cls, v):
    """Convert empty string to None for run_type validation."""
    if v == "":
        return None
    return v

StatisticsIndicator

Bases: BaseModel

Statistics indicator of a run.

You can import the StatisticsIndicator class directly from nextmv:

from nextmv import StatisticsIndicator

PARAMETER	DESCRIPTION
name	Name of the indicator. TYPE: str
value	Value of the indicator. TYPE: Any

Examples:

>>> from nextmv import StatisticsIndicator
>>> indicator = StatisticsIndicator(name="total_cost", value=1250.75)
>>> indicator.name
'total_cost'
>>> indicator.value
1250.75

>>> # Boolean indicator
>>> bool_indicator = StatisticsIndicator(name="optimal", value=True)
>>> bool_indicator.name
'optimal'
>>> bool_indicator.value
True

name instance-attribute

name: str

Name of the indicator.

value instance-attribute

value: Any

Value of the indicator.

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.")