Merge pull request #2 from nextmv-io/feature/eng-4278-create-the-functionality-for-solving-remotely

Author: sebastian-quintero

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/base_model.py

@@ -0,0 +1,20 @@
+"""JSON class for data wrangling JSON objects."""
+
+from typing import Any
+
+from pydantic import BaseModel
+
+
+class BaseModel(BaseModel):
+    """Base class for data wrangling tasks with JSON."""
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]):
+        """Instantiates the class from a dict."""
+
+        return cls(**data)
+
+    def to_dict(self) -> dict[str, Any]:
+        """Converts the class to a dict."""
+
+        return self.model_dump(mode="json", exclude_none=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,136 @@
+"""This module contains the application class."""
+
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+from nextmv.base_model import BaseModel
+from nextmv.cloud.client import Client
+
+
+class Metadata(BaseModel):
+    """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."""
+
+
+class RunResult(BaseModel):
+    """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.
+
+        Returns:
+            ID of the submitted run.
+        """
+
+        payload = {}
+        if input is not None:
+            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,
+        )
+
+        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.
+
+        Returns:
+            Result of the run.
+        """
+
+        response = self.client.get(
+            endpoint=f"{self.endpoint}/runs/{run_id}",
+        )
+
+        return RunResult.from_dict(response.json())

nextmv/cloud/client.py

@@ -0,0 +1,95 @@
+"""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
+    environment variable.
+    """
+
+    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.
+
+        Args:
+            endpoint: Endpoint to send the request to.
+            payload: Payload to send with the request.
+            query_params: Query parameters to send with the request.
+
+        Returns:
+            Response from the Nextmv Cloud API.
+        """
+
+        response = requests.post(
+            url=f"{self.url}/{endpoint}",
+            json=payload,
+            headers=self.headers,
+            params=query_params,
+        )
+        response.raise_for_status()
+
+        return response
+
+    def get(
+        self,
+        endpoint: str,
+        query_params: dict[str, Any] | None = None,
+    ) -> requests.Response:
+        """
+        Send a GET request to the Nextmv Cloud API.
+
+        Args:
+            endpoint: Endpoint to send the request to.
+            query_params: Query parameters to send with the request.
+
+        Returns:
+            Response from the Nextmv Cloud API.
+        """
+
+        response = requests.get(
+            url=f"{self.url}/{endpoint}",
+            headers=self.headers,
+            params=query_params,
+        )
+        response.raise_for_status()
+
+        return response

nextmv/nextroute/schema/base.py

@@ -1,15 +1,13 @@
 """Defines the input class"""
 
-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.base_model import BaseModel
+from nextmv.nextroute.schema.stop import AlternateStop, Stop, StopDefaults
+from nextmv.nextroute.schema.vehicle import Vehicle, VehicleDefaults
 
 
-@dataclass
-class Defaults(_Base):
+class Defaults(BaseModel):
     """Default values for vehicles and stops."""
 
     stops: StopDefaults | None = None
@@ -18,8 +16,7 @@ class Defaults(_Base):
     """Default values for vehicles."""
 
 
-@dataclass
-class DurationGroup(_Base):
+class DurationGroup(BaseModel):
     """Represents a group of stops that get additional duration whenever a stop
     of the group is approached for the first time."""
 
@@ -29,8 +26,7 @@ class DurationGroup(_Base):
     """Stop IDs contained in the group."""
 
 
-@dataclass
-class Input(_Base):
+class Input(BaseModel):
     """Input schema for Nextroute."""
 
     stops: list[Stop]

nextmv/nextroute/schema/input.py

@@ -1,13 +1,10 @@
 """Defines the location class."""
 
 
-from dataclasses import dataclass
+from nextmv.base_model import BaseModel
 
-from .base import _Base
 
-
-@dataclass
-class Location(_Base):
+class Location(BaseModel):
     """Location represents a geographical location."""
 
     lon: float

nextmv/nextroute/schema/location.py

@@ -1,16 +1,14 @@
 """Defines the stop class."""
 
 
-from dataclasses import dataclass
 from datetime import datetime
 from typing import Any
 
-from .base import _Base
-from .location import Location
+from nextmv.base_model import BaseModel
+from nextmv.nextroute.schema.location import Location
 
 
-@dataclass
-class StopDefaults(_Base):
+class StopDefaults(BaseModel):
     """Default values for a stop."""
 
     compatibility_attributes: list[str] | None = None
@@ -33,7 +31,6 @@ class StopDefaults(_Base):
     """Penalty for not planning a stop."""
 
 
-@dataclass(kw_only=True)
 class Stop(StopDefaults):
     """Stop is a location that must be visited by a vehicle in a Vehicle
     Routing Problem (VRP.)"""
@@ -53,7 +50,6 @@ class Stop(StopDefaults):
     """Stops that must be visited before this one on the same route."""
 
 
-@dataclass(kw_only=True)
 class AlternateStop(StopDefaults):
     """An alternate stop can be serviced instead of another stop."""