Applications API Endpoints & Usage

export NEXTMV_API_KEY=YOUR_NEXTMV_CLOUD_API_KEY_HERE

#!/usr/bin/env python3
import datetime
import json
import os
import sys
import time

import requests

NEXTMV_API_KEY = os.environ['NEXTMV_API_KEY']
BASE_URL = 'https://api.cloud.nextmv.io/v1/applications/YOUR_APP_ID'

def solve(data: str) -> dict:
    # Send input to the app and create a run.
    headers = {
        'Authorization': f'Bearer {NEXTMV_API_KEY}',
        'Content-Type': 'application/json',
        'accept': 'application/json',
    }
    params = {'instance_id': 'latest'}
    response = requests.post(
        f'{BASE_URL}/runs',
        data=data,
        headers=headers,
        params=params
    )
    response.raise_for_status()
    run_id = response.json()['run_id']

    # Poll job status until it is finished.
    start = datetime.datetime.now()
    wait_time = 0.2
    wait_max = 5
    timeout = 60
    while True:
        # Check for timeout.
        if (datetime.datetime.now() - start).total_seconds() > timeout:
            raise Exception(f'timed out after {timeout}')

        # Get job status.
        response = requests.get(
            f'{BASE_URL}/runs/{run_id}',
            headers=headers,
        )
        response.raise_for_status()
        status = response.json()['metadata']['status']
        if status == 'running':
            time.sleep(wait_time)
            wait_time = min(wait_time * 2, wait_max)
        elif status == 'succeeded':
            break
        else:
            error = response.json()['metadata']['error']
            raise Exception(f'run failed with status: {status}; error: {error}')

    return response.json()

def main():
    data = {'input': json.load(open(sys.argv[1]))}
    result = solve(json.dumps(data))
    print(json.dumps(result, ensure_ascii=False, indent=4))


if __name__ == '__main__':
    main()

./main.py input.json