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: str | None = 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 | None = None,
    headers: dict[str, str] | None = None,
    max_retries: int = 10,
    status_forcelist: list[int] = (lambda: [429])(),
    timeout: float = 20,
    url: str | None = None,
    console_url: str = "https://cloud.nextmv.io",
    profile: str | None = None,
)

Client that interacts directly with the Nextmv Cloud API.

You can import the Client class directly from cloud:

from nextmv.cloud import Client

The Client class is configured mainly with an API key and an endpoint URL. There are multiple ways to provide these configurations, and the client will check for them in the following order of precedence:

1. The api_key and url attributes set directly on the client instance.
2. The NEXTMV_API_KEY and NEXTMV_ENDPOINT environment variables.
3. If the NEXTMV_PROFILE environment variable is set, it is used to look up the API key and endpoint for that profile in the configuration file (~/.nextmv/config.yaml).
4. If the profile attribute is set on the client, it is used to look up the API key and endpoint for that profile in the configuration file.
5. If the profile attribute is not set, the default profile is used to look up the API key and endpoint in the configuration file.
6. If all of the above lookups fail, an exception is raised indicating that the API key is missing, and the endpoint falls back to the hardcoded default URL of https://api.cloud.nextmv.io.

ATTRIBUTE	DESCRIPTION
api_key	API key to use for authenticating with the Nextmv Cloud API. Resolved from the constructor argument, the NEXTMV_API_KEY environment variable, or the configuration file (in that order). Raises ValueError if no key is found anywhere. TYPE: (str, optional)
allowed_methods	HTTP methods for which failed requests are eligible for retry. Defaults to ["GET", "POST", "PUT", "DELETE"]. Adjust this list to restrict retries to idempotent methods only (e.g., ["GET"]). TYPE: list[str]
backoff_factor	Multiplier applied between retry attempts using exponential backoff. A value of 1 means the successive delays will be 1s, 2s, 4s, … (before jitter). Defaults to 1. TYPE: float
backoff_jitter	Random jitter (in seconds) added to each backoff delay to avoid thundering-herd problems. Defaults to 0.1. TYPE: float
backoff_max	Upper bound on the backoff delay, in seconds. No single wait between retries will exceed this value. Defaults to 60. TYPE: float
configuration_file	Deprecated — no longer used. Kept for backwards compatibility. Previously held the path to the Nextmv CLI configuration file. TYPE: str
headers	HTTP headers sent with every request. Automatically populated during __post_init__ with Authorization: Bearer <api_key> and Content-Type: application/json. Override by passing a custom dict to individual :meth:request calls via the headers parameter. TYPE: (dict[str, str], optional)
max_retries	Total number of retry attempts allowed per request. Defaults to 10. Set to 0 to disable retries. TYPE: int
profile	Named profile to use when looking up credentials in the configuration file (~/.nextmv/config.yaml). Overridden by the NEXTMV_PROFILE environment variable when that variable is set. TYPE: (str, optional)
status_forcelist	HTTP status codes that trigger an automatic retry. Defaults to [429] (Too Many Requests). Add codes such as 500, 502, or 503 to also retry on server errors. TYPE: list[int]
timeout	Maximum number of seconds to wait for the server to send a response. Defaults to 20. Increase for large payloads or slow connections. TYPE: float
url	Base URL of the Nextmv Cloud API. Resolved from the constructor argument, the NEXTMV_ENDPOINT environment variable, or the configuration file (in that order). Defaults to "https://api.cloud.nextmv.io" when not set anywhere. TYPE: (str, optional)
console_url	URL of the Nextmv Cloud web console. Defaults to "https://cloud.nextmv.io". Not used for API requests; provided for convenience when constructing deep-links into the console. TYPE: str

Examples:

Authenticate with an explicit API key:

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

Authenticate via the NEXTMV_API_KEY environment variable (the api_key argument can be omitted):

>>> import os
>>> os.environ["NEXTMV_API_KEY"] = "YOUR_API_KEY"
>>> client = Client()

Use a named profile from the configuration file:

>>> client = Client(profile="staging")

Override the profile with an environment variable:

>>> os.environ["NEXTMV_PROFILE"] = "production"
>>> client = Client()  # uses the "production" profile

Customise retry and timeout behaviour:

>>> client = Client(
...     api_key="YOUR_API_KEY",
...     max_retries=3,
...     backoff_factor=0.5,
...     backoff_jitter=0.05,
...     backoff_max=30,
...     status_forcelist=[429, 500, 502, 503],
...     timeout=60,
... )

Point the client at a self-hosted or staging API endpoint:

>>> client = Client(
...     api_key="YOUR_API_KEY",
...     url="https://staging.api.example.com",
... )

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: str | None = None

API key to use for authenticating with the Nextmv Cloud API.

The API key is determined in the following order of precedence: 1. This api_key attribute set on the client. 2. The NEXTMV_API_KEY environment variable. 3. If the NEXTMV_PROFILE environment variable is set, it is used to look up the API key for that profile in the configuration file. 4. If the profile attribute is set on the client, it is used to look up the API key for that profile in the configuration file. 5. If the profile attribute is not set, the default profile is used to look up the API key in the configuration file. 6. If all of the above lookups fail, an exception is raised indicating that the API key is missing.

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 | None = None

Deprecated. This attribute is no longer being used. Use the profile attribute to specify different configurations instead.

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: dict[str, str] | None = 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.

profile class-attribute instance-attribute

profile: str | None = None

Profile to use from the configuration file. Profiles allow you to configure multiple sets of API keys and endpoints in the configuration file and select between them.

The profile is determined in the following order of precedence: 1. The NEXTMV_PROFILE environment variable. 2. This profile attribute set on the client.

request

request(
    method: str,
    endpoint: str,
    data: Any | None = None,
    headers: dict[str, str] | None = None,
    payload: dict[str, Any] | None = None,
    query_params: dict[str, Any] | None = None,
    json_configurations: dict[str, Any] | None = 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:

List all applications:

>>> client = Client(api_key="YOUR_API_KEY")
>>> response = client.request(method="GET", endpoint="/v1/applications")
>>> print(response.status_code)
200
>>> apps = response.json()
>>> print([a["id"] for a in apps["items"]])
['my-app', 'another-app']

Create a new run with a JSON payload:

>>> run_payload = {
...     "applicationId": "my-app",
...     "instanceId": "candidate",
...     "input": {"value": 10},
... }
>>> response = client.request(
...     method="POST",
...     endpoint="/v1/runs",
...     payload=run_payload,
... )
>>> print(response.json()["id"])
run_xxxxxxxxxxxx

Retrieve a specific run using query parameters:

>>> response = client.request(
...     method="GET",
...     endpoint="/v1/runs",
...     query_params={"applicationId": "my-app", "limit": 5},
... )
>>> print(len(response.json()["items"]))
5

Send a request with custom headers (e.g., to pass a request ID):

>>> response = client.request(
...     method="GET",
...     endpoint="/v1/applications",
...     headers={**client.headers, "X-Request-Id": "abc-123"},
... )

Send a JSON payload with custom serialization (pretty-printed):

>>> response = client.request(
...     method="POST",
...     endpoint="/v1/runs",
...     payload={"applicationId": "my-app", "input": {}},
...     json_configurations={"indent": 2},
... )

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

def request(
    self,
    method: str,
    endpoint: str,
    data: Any | None = None,
    headers: dict[str, str] | None = None,
    payload: dict[str, Any] | None = None,
    query_params: dict[str, Any] | None = None,
    json_configurations: dict[str, Any] | None = 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
    --------
    List all applications:

    >>> client = Client(api_key="YOUR_API_KEY")
    >>> response = client.request(method="GET", endpoint="/v1/applications")
    >>> print(response.status_code)
    200
    >>> apps = response.json()
    >>> print([a["id"] for a in apps["items"]])
    ['my-app', 'another-app']

    Create a new run with a JSON payload:

    >>> run_payload = {
    ...     "applicationId": "my-app",
    ...     "instanceId": "candidate",
    ...     "input": {"value": 10},
    ... }
    >>> response = client.request(
    ...     method="POST",
    ...     endpoint="/v1/runs",
    ...     payload=run_payload,
    ... )
    >>> print(response.json()["id"])
    run_xxxxxxxxxxxx

    Retrieve a specific run using query parameters:

    >>> response = client.request(
    ...     method="GET",
    ...     endpoint="/v1/runs",
    ...     query_params={"applicationId": "my-app", "limit": 5},
    ... )
    >>> print(len(response.json()["items"]))
    5

    Send a request with custom headers (e.g., to pass a request ID):

    >>> response = client.request(
    ...     method="GET",
    ...     endpoint="/v1/applications",
    ...     headers={**client.headers, "X-Request-Id": "abc-123"},
    ... )

    Send a JSON payload with custom serialization (pretty-printed):

    >>> response = client.request(
    ...     method="POST",
    ...     endpoint="/v1/runs",
    ...     payload={"applicationId": "my-app", "input": {}},
    ...     json_configurations={"indent": 2},
    ... )
    """

    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: dict[str, Any] | str | None,
    url: str,
    json_configurations: dict[str, Any] | None = None,
    tar_file: str | None = 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: Union[dict[str, Any], 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
tar_file	If provided, this will be used to upload a tar file instead of a JSON string or dictionary. This is useful for uploading large files that are already packaged as a tarball. If this is provided, data is expected to be None. TYPE: str DEFAULT: None

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

Examples:

Upload a dictionary as JSON (presigned URL obtained from a prior API call):

>>> client = Client(api_key="YOUR_API_KEY")
>>> input_data = {"stops": [{"id": "A"}, {"id": "B"}], "config": {"max_duration": 30}}
>>> client.upload_to_presigned_url(data=input_data, url="PRE_SIGNED_URL")

Upload a raw JSON string:

>>> client.upload_to_presigned_url(
...     data='{"stops": [{"id": "A"}]}',
...     url="PRE_SIGNED_URL",
... )

Upload a pre-built tarball (e.g., a packaged application):

>>> client.upload_to_presigned_url(
...     data=None,
...     url="PRE_SIGNED_URL",
...     tar_file="/path/to/app.tar.gz",
... )

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

def upload_to_presigned_url(
    self,
    data: dict[str, Any] | str | None,
    url: str,
    json_configurations: dict[str, Any] | None = None,
    tar_file: str | None = 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 : Union[dict[str, Any], str], optional
        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.
    tar_file : str, optional
        If provided, this will be used to upload a tar file instead of
        a JSON string or dictionary. This is useful for uploading large
        files that are already packaged as a tarball. If this is provided,
        `data` is expected to be `None`.

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

    Examples
    --------
    Upload a dictionary as JSON (presigned URL obtained from a prior API
    call):

    >>> client = Client(api_key="YOUR_API_KEY")
    >>> input_data = {"stops": [{"id": "A"}, {"id": "B"}], "config": {"max_duration": 30}}
    >>> client.upload_to_presigned_url(data=input_data, url="PRE_SIGNED_URL")  # doctest: +SKIP

    Upload a raw JSON string:

    >>> client.upload_to_presigned_url(  # doctest: +SKIP
    ...     data='{"stops": [{"id": "A"}]}',
    ...     url="PRE_SIGNED_URL",
    ... )

    Upload a pre-built tarball (e.g., a packaged application):

    >>> client.upload_to_presigned_url(  # doctest: +SKIP
    ...     data=None,
    ...     url="PRE_SIGNED_URL",
    ...     tar_file="/path/to/app.tar.gz",
    ... )
    """

    upload_data: str | None = None
    if data is not 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,
    }

    if upload_data is not None:
        kwargs["data"] = upload_data
    elif tar_file is not None and tar_file != "":
        if not os.path.exists(tar_file):
            raise ValueError(f"tar_file {tar_file} does not exist")
        kwargs["data"] = open(tar_file, "rb")
    else:
        raise ValueError("either data or tar_file must be provided")

    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 | None = None

URL (endpoint) of the Nextmv Cloud API.

The endpoint is determined in the following order of precedence: 1. This url attribute set on the client. 2. The NEXTMV_ENDPOINT environment variable. 3. If the NEXTMV_PROFILE environment variable is set, it is used to look up the endpoint for that profile in the configuration file. 4. If the profile attribute is set on the client, it is used to look up the endpoint for that profile in the configuration file. 5. If the profile attribute is not set, the default profile is used to look up the endpoint in the configuration file. 6. If all of the above lookups fail, the hardcoded default URL of https://api.cloud.nextmv.io is used.

get_size

get_size(
    obj: dict[str, Any] | IO[bytes] | str,
    json_configurations: dict[str, Any] | None = 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: dict[str, Any] | IO[bytes] | str, json_configurations: dict[str, Any] | None = 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.")