Client Module

This section documents the client components of the Nextmv Cloud API.

client

Module with the client class.

This module provides the Client class for interacting with the Nextmv Cloud API, and a helper function get_size to determine the size of objects.

CLASS	DESCRIPTION
Client	Client that interacts directly with the Nextmv Cloud API.

FUNCTION	DESCRIPTION
get_size	Finds the size of an object in bytes.

Client dataclass

Client(
    api_key: Optional[str] = None,
    allowed_methods: list[str] = lambda: [
        "GET",
        "POST",
        "PUT",
        "DELETE",
    ](),
    backoff_factor: float = 1,
    backoff_jitter: float = 0.1,
    backoff_max: float = 60,
    configuration_file: str = "~/.nextmv/config.yaml",
    headers: Optional[dict[str, str]] = None,
    max_retries: int = 10,
    status_forcelist: list[int] = lambda: [429](),
    timeout: float = 20,
    url: str = "https://api.cloud.nextmv.io",
    console_url: str = "https://cloud.nextmv.io",
)

Client that interacts directly with the Nextmv Cloud API.

You can import the Client class directly from cloud:

from nextmv.cloud import Client

The API key will be searched, in order of precedence, in:

1. The api_key argument in the constructor.
2. The NEXTMV_API_KEY environment variable.
3. The ~/.nextmv/config.yaml file used by the Nextmv CLI.

PARAMETER	DESCRIPTION
api_key	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. TYPE: str DEFAULT: None
allowed_methods	Allowed HTTP methods to use for retries in requests to the Nextmv Cloud API. Defaults to ["GET", "POST", "PUT", "DELETE"]. TYPE: list[str] DEFAULT: lambda: ['GET', 'POST', 'PUT', 'DELETE']()
backoff_factor	Exponential backoff factor to use for requests to the Nextmv Cloud API. Defaults to 1. TYPE: float DEFAULT: 1
backoff_jitter	Jitter to use for requests to the Nextmv Cloud API when backing off. Defaults to 0.1. TYPE: float DEFAULT: 0.1
backoff_max	Maximum backoff time to use for requests to the Nextmv Cloud API, in seconds. Defaults to 60. TYPE: float DEFAULT: 60
configuration_file	Path to the configuration file used by the Nextmv CLI. Defaults to "~/.nextmv/config.yaml". TYPE: str DEFAULT: '~/.nextmv/config.yaml'
headers	Headers to use for requests to the Nextmv Cloud API. Automatically set up with the API key. TYPE: dict[str, str] DEFAULT: None
max_retries	Maximum number of retries to use for requests to the Nextmv Cloud API. Defaults to 10. TYPE: int DEFAULT: 10
status_forcelist	Status codes to retry for requests to the Nextmv Cloud API. Defaults to [429]. TYPE: list[int] DEFAULT: lambda: [429]()
timeout	Timeout to use for requests to the Nextmv Cloud API, in seconds. Defaults to 20. TYPE: float DEFAULT: 20
url	URL of the Nextmv Cloud API. Defaults to "https://api.cloud.nextmv.io". TYPE: str DEFAULT: 'https://api.cloud.nextmv.io'
console_url	URL of the Nextmv Cloud console. Defaults to "https://cloud.nextmv.io". TYPE: str DEFAULT: 'https://cloud.nextmv.io'

Examples:

>>> client = Client(api_key="YOUR_API_KEY")
>>> response = client.request(method="GET", endpoint="/v1/applications")
>>> print(response.json())

allowed_methods class-attribute instance-attribute

allowed_methods: list[str] = field(
    default_factory=lambda: ["GET", "POST", "PUT", "DELETE"]
)

Allowed HTTP methods to use for retries in requests to the Nextmv Cloud API.

api_key class-attribute instance-attribute

api_key: Optional[str] = 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.

backoff_factor class-attribute instance-attribute

backoff_factor: float = 1

Exponential backoff factor to use for requests to the Nextmv Cloud API.

backoff_jitter class-attribute instance-attribute

backoff_jitter: float = 0.1

Jitter to use for requests to the Nextmv Cloud API when backing off.

backoff_max class-attribute instance-attribute

backoff_max: float = 60

Maximum backoff time to use for requests to the Nextmv Cloud API, in seconds.

configuration_file class-attribute instance-attribute

configuration_file: str = '~/.nextmv/config.yaml'

Path to the configuration file used by the Nextmv CLI.

console_url class-attribute instance-attribute

console_url: str = 'https://cloud.nextmv.io'

URL of the Nextmv Cloud console.

headers class-attribute instance-attribute

headers: Optional[dict[str, str]] = None

Headers to use for requests to the Nextmv Cloud API.

max_retries class-attribute instance-attribute

max_retries: int = 10

Maximum number of retries to use for requests to the Nextmv Cloud API.

request

request(
    method: str,
    endpoint: str,
    data: Optional[Any] = None,
    headers: Optional[dict[str, str]] = None,
    payload: Optional[dict[str, Any]] = None,
    query_params: Optional[dict[str, Any]] = None,
    json_configurations: Optional[dict[str, Any]] = None,
) -> Response

Makes a request to the Nextmv Cloud API.

PARAMETER	DESCRIPTION
method	HTTP method to use (e.g., "GET", "POST"). TYPE: str
endpoint	API endpoint to send the request to (e.g., "/v1/applications"). TYPE: str
data	Data to send in the request body. Typically used for form data. Cannot be used if payload is also provided. TYPE: Any DEFAULT: None
headers	Additional headers to send with the request. These will override the default client headers if keys conflict. TYPE: dict[str, str] DEFAULT: None
payload	JSON payload to send with the request. Prefer using this over data for JSON requests. Cannot be used if data is also provided. TYPE: dict[str, Any] DEFAULT: None
query_params	Query parameters to append to the request URL. TYPE: dict[str, Any] DEFAULT: None
json_configurations	Additional configurations for JSON serialization. This allows customization of the Python json.dumps function, such as specifying indent for pretty printing or default for custom serialization functions. TYPE: dict[str, Any] DEFAULT: None

RETURNS	DESCRIPTION
Response	The response object from the Nextmv Cloud API.

RAISES	DESCRIPTION
HTTPError	If the response status code is not in the 2xx range.
ValueError	If both data and payload are provided. If the payload size exceeds _MAX_LAMBDA_PAYLOAD_SIZE. If the data size exceeds _MAX_LAMBDA_PAYLOAD_SIZE.

Examples:

>>> client = Client(api_key="YOUR_API_KEY")
>>> # Get a list of applications
>>> response = client.request(method="GET", endpoint="/v1/applications")
>>> print(response.status_code)
200
>>> # Create a new run
>>> run_payload = {
...     "applicationId": "app_id",
...     "instanceId": "instance_id",
...     "input": {"value": 10}
... }
>>> response = client.request(
...     method="POST",
...     endpoint="/v1/runs",
...     payload=run_payload
... )
>>> print(response.json()["id"])
run_xxxxxxxxxxxx

Source code in nextmv/nextmv/cloud/client.py

def request(
    self,
    method: str,
    endpoint: str,
    data: Optional[Any] = None,
    headers: Optional[dict[str, str]] = None,
    payload: Optional[dict[str, Any]] = None,
    query_params: Optional[dict[str, Any]] = None,
    json_configurations: Optional[dict[str, Any]] = None,
) -> requests.Response:
    """
    Makes a request to the Nextmv Cloud API.

    Parameters
    ----------
    method : str
        HTTP method to use (e.g., "GET", "POST").
    endpoint : str
        API endpoint to send the request to (e.g., "/v1/applications").
    data : Any, optional
        Data to send in the request body. Typically used for form data.
        Cannot be used if `payload` is also provided.
    headers : dict[str, str], optional
        Additional headers to send with the request. These will override
        the default client headers if keys conflict.
    payload : dict[str, Any], optional
        JSON payload to send with the request. Prefer using this over
        `data` for JSON requests. Cannot be used if `data` is also
        provided.
    query_params : dict[str, Any], optional
        Query parameters to append to the request URL.
    json_configurations : dict[str, Any], optional
        Additional configurations for JSON serialization. This allows
        customization of the Python `json.dumps` function, such as
        specifying `indent` for pretty printing or `default` for custom
        serialization functions.

    Returns
    -------
    requests.Response
        The response object from the Nextmv Cloud API.

    Raises
    ------
    requests.HTTPError
        If the response status code is not in the 2xx range.
    ValueError
        If both `data` and `payload` are provided.
        If the `payload` size exceeds `_MAX_LAMBDA_PAYLOAD_SIZE`.
        If the `data` size exceeds `_MAX_LAMBDA_PAYLOAD_SIZE`.

    Examples
    --------
    >>> client = Client(api_key="YOUR_API_KEY")
    >>> # Get a list of applications
    >>> response = client.request(method="GET", endpoint="/v1/applications")
    >>> print(response.status_code)
    200
    >>> # Create a new run
    >>> run_payload = {
    ...     "applicationId": "app_id",
    ...     "instanceId": "instance_id",
    ...     "input": {"value": 10}
    ... }
    >>> response = client.request(
    ...     method="POST",
    ...     endpoint="/v1/runs",
    ...     payload=run_payload
    ... )
    >>> print(response.json()["id"])
    run_xxxxxxxxxxxx
    """

    if payload is not None and data is not None:
        raise ValueError("cannot use both data and payload")

    if (
        payload is not None
        and get_size(payload, json_configurations=json_configurations) > _MAX_LAMBDA_PAYLOAD_SIZE
    ):
        raise ValueError(
            f"payload size of {get_size(payload, json_configurations=json_configurations)} bytes exceeds "
            + f"the maximum allowed size of {_MAX_LAMBDA_PAYLOAD_SIZE} bytes"
        )

    if data is not None and get_size(data, json_configurations=json_configurations) > _MAX_LAMBDA_PAYLOAD_SIZE:
        raise ValueError(
            f"data size of {get_size(data, json_configurations=json_configurations)} bytes exceeds "
            + f"the maximum allowed size of {_MAX_LAMBDA_PAYLOAD_SIZE} bytes"
        )

    session = requests.Session()
    retries = Retry(
        total=self.max_retries,
        backoff_factor=self.backoff_factor,
        backoff_jitter=self.backoff_jitter,
        backoff_max=self.backoff_max,
        status_forcelist=self.status_forcelist,
        allowed_methods=self.allowed_methods,
    )
    adapter = HTTPAdapter(max_retries=retries)
    session.mount("https://", adapter)

    kwargs: dict[str, Any] = {
        "url": urljoin(self.url, endpoint),
        "timeout": self.timeout,
    }
    kwargs["headers"] = headers if headers is not None else self.headers
    if data is not None:
        kwargs["data"] = data
    if payload is not None:
        if isinstance(payload, (dict, list)):
            data = deflated_serialize_json(payload, json_configurations=json_configurations)
            kwargs["data"] = data
        else:
            raise ValueError("payload must be a dictionary or a list")
    if query_params is not None:
        kwargs["params"] = query_params

    response = session.request(method=method, **kwargs)

    try:
        response.raise_for_status()
    except requests.HTTPError as e:
        raise requests.HTTPError(
            f"request to {endpoint} failed with status code {response.status_code} and message: {response.text}"
        ) from e

    return response

status_forcelist class-attribute instance-attribute

status_forcelist: list[int] = field(
    default_factory=lambda: [429]
)

Status codes to retry for requests to the Nextmv Cloud API.

timeout class-attribute instance-attribute

timeout: float = 20

Timeout to use for requests to the Nextmv Cloud API.

upload_to_presigned_url

upload_to_presigned_url(
    data: Union[dict[str, Any], str],
    url: str,
    json_configurations: Optional[dict[str, Any]] = None,
) -> None

Uploads data to a presigned URL.

This method is typically used for uploading large input or output files directly to cloud storage, bypassing the main API for efficiency.

PARAMETER	DESCRIPTION
data	The data to upload. If a dictionary is provided, it will be JSON-serialized. If a string is provided, it will be uploaded as is. TYPE: dict[str, Any] or str
url	The presigned URL to which the data will be uploaded. TYPE: str
json_configurations	Additional configurations for JSON serialization. This allows customization of the Python json.dumps function, such as specifying indent for pretty printing or default for custom serialization functions. TYPE: dict[str, Any] DEFAULT: None

RAISES	DESCRIPTION
ValueError	If data is not a dictionary or a string.
HTTPError	If the upload request fails.

Examples:

Assume presigned_upload_url is obtained from a previous API call.

>>> client = Client(api_key="YOUR_API_KEY")
>>> input_data = {"value": 42, "items": [1, 2, 3]}
>>> client.upload_to_presigned_url(data=input_data, url="PRE_SIGNED_URL")

Source code in nextmv/nextmv/cloud/client.py

def upload_to_presigned_url(
    self, data: Union[dict[str, Any], str], url: str, json_configurations: Optional[dict[str, Any]] = None
) -> None:
    """
    Uploads data to a presigned URL.

    This method is typically used for uploading large input or output files
    directly to cloud storage, bypassing the main API for efficiency.

    Parameters
    ----------
    data : dict[str, Any] or str
        The data to upload. If a dictionary is provided, it will be
        JSON-serialized. If a string is provided, it will be uploaded
        as is.
    url : str
        The presigned URL to which the data will be uploaded.
    json_configurations : dict[str, Any], optional
        Additional configurations for JSON serialization. This allows
        customization of the Python `json.dumps` function, such as
        specifying `indent` for pretty printing or `default` for custom
        serialization functions.

    Raises
    ------
    ValueError
        If `data` is not a dictionary or a string.
    requests.HTTPError
        If the upload request fails.

    Examples
    --------
    Assume `presigned_upload_url` is obtained from a previous API call.
    >>> client = Client(api_key="YOUR_API_KEY")
    >>> input_data = {"value": 42, "items": [1, 2, 3]}
    >>> client.upload_to_presigned_url(data=input_data, url="PRE_SIGNED_URL") # doctest: +SKIP
    """

    upload_data: Optional[str] = None
    if isinstance(data, dict):
        upload_data = deflated_serialize_json(data, json_configurations=json_configurations)
    elif isinstance(data, str):
        upload_data = data
    else:
        raise ValueError("data must be a dictionary or a string")

    session = requests.Session()
    retries = Retry(
        total=self.max_retries,
        backoff_factor=self.backoff_factor,
        backoff_jitter=self.backoff_jitter,
        backoff_max=self.backoff_max,
        status_forcelist=self.status_forcelist,
        allowed_methods=self.allowed_methods,
    )
    adapter = HTTPAdapter(max_retries=retries)
    session.mount("https://", adapter)
    kwargs: dict[str, Any] = {
        "url": url,
        "timeout": self.timeout,
        "data": upload_data,
    }

    response = session.put(**kwargs)

    try:
        response.raise_for_status()
    except requests.HTTPError as e:
        raise requests.HTTPError(
            f"upload to presigned URL {url} failed with "
            f"status code {response.status_code} and message: {response.text}"
        ) from e

url class-attribute instance-attribute

url: str = 'https://api.cloud.nextmv.io'

URL of the Nextmv Cloud API.

get_size

get_size(
    obj: Union[dict[str, Any], IO[bytes], str],
    json_configurations: Optional[dict[str, Any]] = None,
) -> int

Finds the size of an object in bytes.

This function supports dictionaries (JSON-serialized), file-like objects (by reading their content), and strings.

PARAMETER	DESCRIPTION
obj	The object whose size is to be determined. - If a dict, it's converted to a JSON string. - If a file-like object (e.g., opened file), its size is read. - If a string, its UTF-8 encoded byte length is calculated. TYPE: dict[str, Any] or IO[bytes] or str
json_configurations	Additional configurations for JSON serialization. This allows customization of the Python json.dumps function, such as specifying indent for pretty printing or default for custom serialization functions. TYPE: dict[str, Any] DEFAULT: None

RETURNS	DESCRIPTION
int	The size of the object in bytes.

RAISES	DESCRIPTION
TypeError	If the object type is not supported (i.e., not a dict, file-like object, or string).

Examples:

>>> my_dict = {"key": "value", "number": 123}
>>> get_size(my_dict)
30
>>> import io
>>> my_string = "Hello, Nextmv!"
>>> string_io = io.StringIO(my_string)
>>> # To get size of underlying buffer for StringIO, we need to encode
>>> string_bytes_io = io.BytesIO(my_string.encode('utf-8'))
>>> get_size(string_bytes_io)
14
>>> get_size("Hello, Nextmv!")
14

Source code in nextmv/nextmv/cloud/client.py

def get_size(obj: Union[dict[str, Any], IO[bytes], str], json_configurations: Optional[dict[str, Any]] = None) -> int:
    """
    Finds the size of an object in bytes.

    This function supports dictionaries (JSON-serialized), file-like objects
    (by reading their content), and strings.

    Parameters
    ----------
    obj : dict[str, Any] or IO[bytes] or str
        The object whose size is to be determined.
        - If a dict, it's converted to a JSON string.
        - If a file-like object (e.g., opened file), its size is read.
        - If a string, its UTF-8 encoded byte length is calculated.
    json_configurations : dict[str, Any], optional
        Additional configurations for JSON serialization. This allows
        customization of the Python `json.dumps` function, such as specifying
        `indent` for pretty printing or `default` for custom serialization
        functions.

    Returns
    -------
    int
        The size of the object in bytes.

    Raises
    ------
    TypeError
        If the object type is not supported (i.e., not a dict,
        file-like object, or string).

    Examples
    --------
    >>> my_dict = {"key": "value", "number": 123}
    >>> get_size(my_dict)
    30
    >>> import io
    >>> my_string = "Hello, Nextmv!"
    >>> string_io = io.StringIO(my_string)
    >>> # To get size of underlying buffer for StringIO, we need to encode
    >>> string_bytes_io = io.BytesIO(my_string.encode('utf-8'))
    >>> get_size(string_bytes_io)
    14
    >>> get_size("Hello, Nextmv!")
    14
    """

    if isinstance(obj, dict):
        obj_str = deflated_serialize_json(obj, json_configurations=json_configurations)
        return len(obj_str.encode("utf-8"))

    elif hasattr(obj, "read"):
        obj.seek(0, 2)  # Move the cursor to the end of the file
        size = obj.tell()
        obj.seek(0)  # Reset the cursor to the beginning of the file
        return size

    elif isinstance(obj, str):
        return len(obj.encode("utf-8"))

    else:
        raise TypeError("Unsupported type. Only dictionaries, file objects (IO[bytes]), and strings are supported.")