Python Sdk

Python SDK

Reference for running an app remotely with the Python SDK.

Export your API key as an environment variable:

export NEXTMV_API_KEY="<YOUR-API-KEY>"
Copy

Complete run workflow

Execute the complete run workflow (submit a run, poll for status, get the results). The output key will contain the actual output of the run. This is an aggregation of the other steps.

import json
import os

from nextmv.cloud import Application, Client, PollingOptions

with open(os.getenv("INPUT_FILE")) as f:
    input = json.load(f)

client = Client(api_key=os.getenv("NEXTMV_API_KEY"))
app = Application(client=client, id=os.getenv("APP_ID"))
result = app.new_run_with_result(
    input=input,
    instance_id=os.getenv("INSTANCE_ID"),
    run_options={
        "solve.duration": "10s",
        "solve.iterations": "20",
    },
    polling_options=PollingOptions(),  # Customize polling options.
)
print(json.dumps(result.to_dict(), indent=2))  # Pretty print.
Copy

New run

Submit a new run. The run_id is returned.

import json
import os

from nextmv.cloud import Application, Client

with open(os.getenv("INPUT_FILE")) as f:
    input = json.load(f)

client = Client(api_key=os.getenv("NEXTMV_API_KEY"))
app = Application(client=client, id=os.getenv("APP_ID"))
run_id = app.new_run(
    input=input,
    instance_id=os.getenv("INSTANCE_ID"),
    options={
        "solve.duration": "10s",
        "solve.iterations": "20",
    },
)
print(run_id)
Copy

Run metadata

Get a run metadata. The metadata.status key will contain the status of the run. Use the run_id obtained from submitting a run.

import json
import os

from nextmv.cloud import Application, Client

client = Client(api_key=os.getenv("NEXTMV_API_KEY"))
app = Application(client=client, id=os.getenv("APP_ID"))
metadata = app.run_metadata(run_id=os.getenv("RUN_ID"))
print(json.dumps(metadata.to_dict(), indent=2))  # Pretty print.
Copy

Run result

Get a run result. The output key will contain the actual output of the run. Use the run_id obtained from submitting a run.

import json
import os

from nextmv.cloud import Application, Client

client = Client(api_key=os.getenv("NEXTMV_API_KEY"))
app = Application(client=client, id=os.getenv("APP_ID"))
result = app.run_result(run_id=os.getenv("RUN_ID"))
print(json.dumps(result.to_dict(), indent=2))  # Pretty print.
Copy

The app.run_result_with_polling method is also available. It will poll for the run status and return the result when it is succeeded.

Cancel a run

To cancel a run, use the cancel_run method.

import os

from nextmv.cloud import Application, Client

client = Client(api_key=os.getenv("NEXTMV_API_KEY"))
app = Application(client=client, id="<YOUR-APP-ID>")
app.cancel_run(run_id="<YOUR-RUN-ID>")
Copy

Account queue

Get the queue of runs for the account. A run is queued when the status_v2 in the metadata is queued. Use the Account.queue method.

import json
import os

from nextmv.cloud import Client
from nextmv.cloud.account import Account

client = Client(api_key=os.getenv("NEXTMV_API_KEY_PROD"))
account = Account(client=client)
queue = account.queue()

print(json.dumps(queue.to_dict(), indent=2))  # Pretty print.
Copy

Page last updated

Go to on-page nav menu