Add the cloud.client and cloud.application logic

nextmv-py.code-workspace

@@ -5,7 +5,7 @@
     }
   ],
   "settings": {
-    "python.testing.unittestArgs": ["-v", "-s", "./tests", "-p", "test*.py"],
+    "python.testing.unittestArgs": ["-v", "-s", ".", "-p", "test*.py"],
     "python.testing.pytestEnabled": false,
     "python.testing.unittestEnabled": true
   }

nextmv/cloud/__init__.py

@@ -0,0 +1,4 @@
+"""Functionality for interacting with the Nextmv Cloud."""
+
+from .application import Application as Application
+from .client import Client as Client

nextmv/cloud/application.py

@@ -0,0 +1,134 @@
+"""This module contains the application class."""
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+from nextmv.cloud.client import Client
+from nextmv.json_class import JSONClass
+
+
+@dataclass
+class Metadata(JSONClass):
+    """Metadata of a run, whether it was successful or not."""
+
+    status: str
+    """Status of the run."""
+    created_at: datetime
+    """Date and time when the run was created."""
+    duration: float
+    """Duration of the run in milliseconds."""
+    input_size: float
+    """Size of the input in bytes."""
+    output_size: float
+    """Size of the output in bytes."""
+    error: str
+    """Error message if the run failed."""
+    application_id: str
+    """ID of the application where the run was submitted to."""
+    application_instance_id: str
+    """ID of the instance where the run was submitted to."""
+    application_version_id: str
+    """ID of the version of the application where the run was submitted to."""
+
+
+@dataclass
+class RunResult(JSONClass):
+    """Result of a run, wheter it was successful or not."""
+
+    id: str
+    """ID of the run."""
+    user_email: str
+    """Email of the user who submitted the run."""
+    name: str
+    """Name of the run."""
+    description: str
+    """Description of the run."""
+    metadata: Metadata
+    """Metadata of the run."""
+    output: dict[str, Any]
+    """Output of the run."""
+
+
+@dataclass
+class Application:
+    """An application is a published decision model that can be executed."""
+
+    client: Client
+    """Client to use for interacting with the Nextmv Cloud API."""
+    id: str
+    """ID of the application."""
+    endpoint: str = "v1/applications/{id}"
+    """Base endpoint for the application."""
+    default_instance_id: str = "devint"
+    """Default instance ID to use for submitting runs."""
+
+    def __post_init__(self):
+        """Logic to run after the class is initialized."""
+
+        self.endpoint = self.endpoint.format(id=self.id)
+
+    def new_run(
+        self,
+        input: dict[str, Any] = None,
+        instance_id: str | None = None,
+        name: str | None = None,
+        description: str | None = None,
+        upload_id: str | None = None,
+        options: dict[str, Any] | None = None,
+    ) -> str:
+        """
+        Submit an input to start a new run of the application. Returns the
+        run_id of the submitted run.
+
+        Args:
+            input: Input to use for the run.
+            instance_id: ID of the instance to use for the run. If not
+                provided, the default_instance_id will be used.
+            name: Name of the run.
+            description: Description of the run.
+            upload_id: ID to use when running a large input.
+            options: Options to use for the run.
+        """
+
+        payload = {
+            "input": input,
+        }
+        if name is not None:
+            payload["name"] = name
+        if description is not None:
+            payload["description"] = description
+        if upload_id is not None:
+            payload["upload_id"] = upload_id
+        if options is not None:
+            payload["options"] = options
+
+        query_params = {
+            "instance_id": instance_id if instance_id is not None else self.default_instance_id,
+        }
+        response = self.client.post(
+            endpoint=f"{self.endpoint}/runs",
+            payload=payload,
+            query_params=query_params,
+        )
+        response.raise_for_status()
+
+        return response.json()["run_id"]
+
+    def run_result(
+        self,
+        run_id: str,
+    ) -> RunResult:
+        """
+        Get the result of a run.
+
+        Args:
+            run_id: ID of the run.
+        """
+
+        response = self.client.get(
+            endpoint=f"{self.endpoint}/runs/{run_id}",
+        )
+        response.raise_for_status()
+
+        return RunResult.from_dict(response.json())

nextmv/cloud/client.py

@@ -0,0 +1,67 @@
+"""Module with the client class."""
+
+import os
+from dataclasses import dataclass
+from typing import Any
+
+import requests
+
+
+@dataclass
+class Client:
+    """Client that interacts directly with the Nextmv Cloud API. The API key
+    must be provided either in the constructor or via the NEXTMV_API_KEY"""
+
+    api_key: str | None = None
+    """API key to use for authenticating with the Nextmv Cloud API. If not
+    provided, the client will look for the NEXTMV_API_KEY environment
+    variable."""
+    url: str = "https://api.cloud.nextmv.io"
+    """URL of the Nextmv Cloud API."""
+    headers: dict[str, str] | None = None
+    """Headers to use for requests to the Nextmv Cloud API."""
+
+    def __post_init__(self):
+        """Logic to run after the class is initialized."""
+
+        if self.api_key is None:
+            api_key = os.getenv("NEXTMV_API_KEY")
+            if api_key is None:
+                raise ValueError(
+                    "no API key provided. Either set it in the constructor or "
+                    "set the NEXTMV_API_KEY environment variable."
+                )
+            self.api_key = api_key
+
+        self.headers = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json",
+        }
+
+    def post(
+        self,
+        endpoint: str,
+        payload: dict[str, Any],
+        query_params: dict[str, Any] | None = None,
+    ) -> requests.Response:
+        """Send a POST request to the Nextmv Cloud API."""
+
+        return requests.post(
+            url=f"{self.url}/{endpoint}",
+            json=payload,
+            headers=self.headers,
+            params=query_params,
+        )
+
+    def get(
+        self,
+        endpoint: str,
+        query_params: dict[str, Any] | None = None,
+    ) -> requests.Response:
+        """Send a GET request to the Nextmv Cloud API."""
+
+        return requests.get(
+            url=f"{self.url}/{endpoint}",
+            headers=self.headers,
+            params=query_params,
+        )

nextmv/nextroute/schema/base.py → nextmv/json_class.py

@@ -1,12 +1,12 @@
-"""Base class for data wrangling."""
+"""JSON class for data wrangling JSON objects."""
 
 import json
 from dataclasses import asdict, dataclass
 from typing import Any
 
 
 @dataclass
-class _Base:
+class JSONClass:
     """Base class for data wrangling tasks."""
 
     @classmethod
@@ -16,7 +16,7 @@ def from_dict(cls, data: dict[str, Any]):
         return cls(**data)
 
     @classmethod
-    def from_json(cls, filepath: str):
+    def from_file(cls, filepath: str):
         """Instantiates the class from a JSON file."""
 
         with open(filepath) as f:

nextmv/nextroute/schema/input.py

@@ -3,13 +3,13 @@
 from dataclasses import dataclass
 from typing import Any
 
-from .base import _Base
-from .stop import AlternateStop, Stop, StopDefaults
-from .vehicle import Vehicle, VehicleDefaults
+from nextmv.json_class import JSONClass
+from nextmv.nextroute.schema.stop import AlternateStop, Stop, StopDefaults
+from nextmv.nextroute.schema.vehicle import Vehicle, VehicleDefaults
 
 
 @dataclass
-class Defaults(_Base):
+class Defaults(JSONClass):
     """Default values for vehicles and stops."""
 
     stops: StopDefaults | None = None
@@ -19,7 +19,7 @@ class Defaults(_Base):
 
 
 @dataclass
-class DurationGroup(_Base):
+class DurationGroup(JSONClass):
     """Represents a group of stops that get additional duration whenever a stop
     of the group is approached for the first time."""
 
@@ -30,7 +30,7 @@ class DurationGroup(_Base):
 
 
 @dataclass
-class Input(_Base):
+class Input(JSONClass):
     """Input schema for Nextroute."""
 
     stops: list[Stop]

nextmv/nextroute/schema/location.py

@@ -3,11 +3,11 @@
 
 from dataclasses import dataclass
 
-from .base import _Base
+from nextmv.json_class import JSONClass
 
 
 @dataclass
-class Location(_Base):
+class Location(JSONClass):
     """Location represents a geographical location."""
 
     lon: float

nextmv/nextroute/schema/stop.py

@@ -5,12 +5,12 @@
 from datetime import datetime
 from typing import Any
 
-from .base import _Base
-from .location import Location
+from nextmv.json_class import JSONClass
+from nextmv.nextroute.schema.location import Location
 
 
 @dataclass
-class StopDefaults(_Base):
+class StopDefaults(JSONClass):
     """Default values for a stop."""
 
     compatibility_attributes: list[str] | None = None

nextmv/nextroute/schema/vehicle.py

@@ -5,12 +5,12 @@
 from datetime import datetime
 from typing import Any
 
-from .base import _Base
-from .location import Location
+from nextmv.json_class import JSONClass
+from nextmv.nextroute.schema.location import Location
 
 
 @dataclass
-class InitialStop(_Base):
+class InitialStop(JSONClass):
     """Represents a stop that is already planned on a vehicle."""
 
     id: str
@@ -21,7 +21,7 @@ class InitialStop(_Base):
 
 
 @dataclass
-class VehicleDefaults(_Base):
+class VehicleDefaults(JSONClass):
     """Default values for vehicles."""
 
     activation_penalty: int | None = None

requirements.txt

@@ -1,3 +1,4 @@
+requests==2.31.0
 ruff==0.1.7
 setuptools==69.0.2
 wheel==0.42.0

tests/nextroute/schema/test_input.py

@@ -2,13 +2,13 @@
 import os
 import unittest
 
-from nextmv.nextroute.schema import Input
+from nextmv.nextroute.schema import Input, Stop
 
 
 class TestInput(unittest.TestCase):
     def test_from_json(self):
         filepath = os.path.join(os.path.dirname(__file__), "input.json")
-        input = Input.from_json(filepath)
+        input = Input.from_file(filepath)
         parsed = input.to_dict()
         with open(filepath) as f:
             expected = json.load(f)
@@ -19,3 +19,17 @@ def test_from_json(self):
             expected,
             "Parsing the JSON into the class and back should yield the same result.",
         )
+
+    def test_from_dict(self):
+        filepath = os.path.join(os.path.dirname(__file__), "input.json")
+        with open(filepath) as f:
+            expected = json.load(f)
+
+        input = Input.from_dict(expected)
+        stops = input.stops
+        for stop in stops:
+            self.assertTrue(isinstance(stop, Stop), f"Stop {stop} should be of type Stop.")
+
+        vehicles = input.vehicles
+        for vehicle in vehicles:
+            self.assertTrue(isinstance(vehicle, Stop), f"Vehicle {vehicle} should be of type Vehicle.")