Managing queuing¶

Info

Learn the concepts and fundamentals of queuing in the Explanation page.

Queuing lets you submit runs beyond your allowed maximum concurrency limits. Queues are maintained for each execution class and ordered by priority then age. Here are the controls you have for queuing:

Setting the priority, which must be a number between 1 (highest) and 9 (lowest). Default is 6.
Disabling queuing. Default behavior is to queue.

You can customize queuing behavior impacting runs by modifying the instance configuration, or directly when starting a run.

Managing queuing when running¶

You can configure queuing behavior with the Application.new_run_with_result method (or Application.new_run) in two ways:

Use directly when starting a run with the queuing attribute of a RunConfiguration object, passing a RunQueuing object, or
Attach it to an instance configuration with the queuing attribute of an InstanceConfiguration object and use that instance when starting a run.

Here is an example where the queuing behavior is modified when starting a run.

main.pyBashOutput

import os

import nextmv
from nextmv import cloud
from nextmv import RunConfiguration, RunQueuing

client = cloud.Client(api_key=os.getenv("NEXTMV_API_KEY"))
app = cloud.Application.get(client=client, id="rustling-lynx")
run_result = app.new_run_with_result(
    input={"name": "world", "radius": 6378, "distance": 147.6},
    instance_id="latest",
    configuration=RunConfiguration(queuing=RunQueuing(priority=2)),
)

result = run_result.to_dict()
result["output"].pop("assets", None)  # Assets are omitted here for a cleaner display.

nextmv.write(result)

uv run main.py

{
  "description": "",
  "id": "latest-WeGfXyEDR",
  "metadata": {
    "application_id": "rustling-lynx",
    "application_instance_id": "latest",
    "application_version_id": "",
    "created_at": "2026-07-28T19:31:29Z",
    "duration": 6012.0,
    "error": "",
    "execution_class": "6c9500mb870s",
    "execution_duration": 5688.0,
    "format": {
      "input": {
        "type": "json"
      },
      "output": {
        "type": "json"
      }
    },
    "initiated_at": "2026-07-28T19:31:29.672444Z",
    "input_size": 47.0,
    "metrics": {
      "message": "Hello, world",
      "value": 1.23
    },
    "options": {
      "active_options": {
        "details": "true"
      },
      "options_summary": [
        {
          "name": "details",
          "source": "version",
          "value": "true"
        }
      ]
    },
    "output_size": 25099.0,
    "queuing_disabled": false,
    "queuing_priority": 2,
    "run_type": {
      "type": "standard",
      "definition_id": "",
      "reference_id": ""
    },
    "runtime": "python-3_11",
    "status_v2": "succeeded"
  },
  "name": "",
  "user_email": "sebastian@nextmv.io",
  "console_url": "https://cloud.nextmv.io/app/rustling-lynx/run/latest-WeGfXyEDR?view=details",
  "output": {
    "options": {
      "details": true
    },
    "solution": {
      "message": "Hello, world"
    },
    "metrics": {
      "value": 1.23,
      "message": "Hello, world"
    }
  }
}

Let's say we have an instance identified as production. First, let's update it to disable queuing with the Application.update_instance method.

main.pyBashOutput

import os

import nextmv
from nextmv import cloud
from nextmv import RunQueuing

client = cloud.Client(api_key=os.getenv("NEXTMV_API_KEY"))
app = cloud.Application.get(client=client, id="rustling-lynx")
instance = app.update_instance(
    id="production",
    configuration=cloud.InstanceConfiguration(queuing=RunQueuing(disabled=True)),
)

nextmv.write(instance.to_dict())

uv run main.py

{
  "id": "production",
  "application_id": "rustling-lynx",
  "version_id": "v0.0.1",
  "name": "The main production instance",
  "description": "",
  "configuration": {
    "execution_class": "6c9500mb870s",
    "queuing": {
      "priority": 6,
      "disabled": true
    }
  },
  "locked": false,
  "created_at": "2026-07-28T19:31:20.632023Z",
  "updated_at": "2026-07-28T19:31:47.692455Z"
}

We can now start a run using the production instance, and the attached queuing configuration will be used automatically.

main.pyBashOutput

import os

import nextmv
from nextmv import cloud

client = cloud.Client(api_key=os.getenv("NEXTMV_API_KEY"))
app = cloud.Application.get(client=client, id="rustling-lynx")
run_result = app.new_run_with_result(
    input={"name": "world", "radius": 6378, "distance": 147.6},
    instance_id="production",
)

result = run_result.to_dict()
result["output"].pop("assets", None)  # Assets are omitted here for a cleaner display.

nextmv.write(result)

uv run main.py

{
  "description": "",
  "id": "production-71eBuyPvg",
  "metadata": {
    "application_id": "rustling-lynx",
    "application_instance_id": "production",
    "application_version_id": "v0.0.1",
    "created_at": "2026-07-28T19:31:56Z",
    "duration": 5375.0,
    "error": "",
    "execution_class": "6c9500mb870s",
    "execution_duration": 5121.0,
    "format": {
      "input": {
        "type": "json"
      },
      "output": {
        "type": "json"
      }
    },
    "initiated_at": "2026-07-28T19:31:56.818265Z",
    "input_size": 47.0,
    "metrics": {
      "message": "Hello, world",
      "value": 1.23
    },
    "options": {
      "active_options": {
        "details": "true"
      },
      "options_summary": [
        {
          "name": "details",
          "source": "version",
          "value": "true"
        }
      ]
    },
    "output_size": 25099.0,
    "queuing_disabled": true,
    "queuing_priority": 6,
    "run_type": {
      "type": "standard",
      "definition_id": "",
      "reference_id": ""
    },
    "runtime": "python-3_11",
    "status_v2": "succeeded"
  },
  "name": "",
  "user_email": "sebastian@nextmv.io",
  "console_url": "https://cloud.nextmv.io/app/rustling-lynx/run/production-71eBuyPvg?view=details",
  "output": {
    "options": {
      "details": true
    },
    "solution": {
      "message": "Hello, world"
    },
    "metrics": {
      "value": 1.23,
      "message": "Hello, world"
    }
  }
}

Getting the queue of runs¶

You can use the Application.list_runs method with the status keyword argument set to StatusV2.queued to retrieve the list of runs that are currently queued.

main.pyBashOutput

import json
import os

from nextmv import cloud
from nextmv import StatusV2

client = cloud.Client(api_key=os.getenv("NEXTMV_API_KEY"))
app = cloud.Application.get(client=client, id="rustling-lynx")
runs = app.list_runs(status=StatusV2.queued)

print(json.dumps([run.to_dict() for run in runs], indent=2))

uv run main.py

[
  {
    "id": "latest-isV8vKEDR",
    "user_email": "sebastian@nextmv.io",
    "name": "",
    "description": "",
    "created_at": "2026-07-23T22:28:14.690523Z",
    "application_id": "rustling-lynx",
    "application_instance_id": "latest",
    "application_version_id": "",
    "run_type": {
      "type": "standard",
      "definition_id": "",
      "reference_id": ""
    },
    "execution_class": "6c9500mb870s",
    "runtime": "python-3_11",
    "status_v2": "queued",
    "queuing_priority": 6,
    "queuing_disabled": false,
    "options": {
      "details": "true"
    },
    "options_summary": [
      {
        "name": "details",
        "value": "true",
        "source": "version"
      }
    ]
  }
]