Integration Module

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

integration

Integration module for interacting with Nextmv Cloud integrations.

This module provides functionality to interact with integrations in Nextmv Cloud, including integration management.

CLASS	DESCRIPTION
IntegrationType	Enum representing the type of an integration.
IntegrationProvider	Enum representing the provider of an integration.
Integration	Class representing an integration in Nextmv Cloud.

FUNCTION	DESCRIPTION
list_integrations	Function to list all integrations in Nextmv Cloud.

Integration

Bases: BaseModel

Represents an integration in Nextmv Cloud. An integration allows Nextmv Cloud to communicate with external systems or services.

You can import the Integration class directly from cloud:

from nextmv.cloud import Integration

You can use the Integration.get class method to retrieve an existing integration from Nextmv Cloud, to ensure that all fields are properly populated.

PARAMETER	DESCRIPTION
integration_id	The unique identifier of the integration. TYPE: str
client	Client to use for interacting with the Nextmv Cloud API. TYPE: Client
name	The name of the integration. TYPE: str
description	An optional description of the integration. TYPE: str
is_global	Indicates whether the integration is global (available to all applications in the account). TYPE: bool
application_ids	List of application IDs that have access to this integration. TYPE: list[str]
integration_type	The type of the integration (runtime or data). TYPE: IntegrationType
exec_types	List of execution types supported by the integration. TYPE: list[ManifestType]
provider	The provider of the integration. TYPE: IntegrationProvider
provider_config	Configuration specific to the integration provider. TYPE: dict[str, Any]
created_at	The timestamp when the integration was created. TYPE: datetime
updated_at	The timestamp when the integration was last updated. TYPE: datetime

application_ids class-attribute instance-attribute

application_ids: list[str] | None = None

List of application IDs that have access to this integration.

client class-attribute instance-attribute

client: Client = Field(exclude=True)

Client to use for interacting with the Nextmv Cloud API.

created_at class-attribute instance-attribute

created_at: datetime | None = None

The timestamp when the integration was created.

delete

delete() -> None

Deletes the integration from Nextmv Cloud.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> from nextmv.cloud import Client, Integration
>>> client = Client(api_key="your_api_key")
>>> integration = Integration.get(client=client, integration_id="your_integration_id")
>>> integration.delete()

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

def delete(self) -> None:
    """
    Deletes the integration from Nextmv Cloud.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> from nextmv.cloud import Client, Integration
    >>> client = Client(api_key="your_api_key")
    >>> integration = Integration.get(client=client, integration_id="your_integration_id")
    >>> integration.delete()
    """

    _ = self.client.request(
        method="DELETE",
        endpoint=self.endpoint,
    )

description class-attribute instance-attribute

description: str | None = None

An optional description of the integration.

endpoint class-attribute instance-attribute

endpoint: str = Field(
    exclude=True, default="v1/integrations/{id}"
)

Base endpoint for the integration.

exec_types class-attribute instance-attribute

exec_types: list[ManifestType] | None = None

List of execution types supported by the integration.

get classmethod

get(client: Client, integration_id: str) -> Integration

Retrieve an existing integration from Nextmv Cloud.

This method should be used for validating that the integration exists, and not rely simply on instantiating the Integration class. Using this method ensures that all the fields of the Integration class are properly populated.

PARAMETER	DESCRIPTION
client	Client to use for interacting with the Nextmv Cloud API. TYPE: Client
integration_id	The unique identifier of the integration to retrieve. TYPE: str

RETURNS	DESCRIPTION
Integration	The retrieved integration instance.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> from nextmv.cloud import Client, Integration
>>> client = Client(api_key="your_api_key")
>>> integration = Integration.get(client=client, integration_id="your_integration_id")
>>> print(integration.to_dict())

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

@classmethod
def get(cls, client: Client, integration_id: str) -> "Integration":
    """
    Retrieve an existing integration from Nextmv Cloud.

    This method should be used for validating that the integration exists,
    and not rely simply on instantiating the `Integration` class. Using
    this method ensures that all the fields of the `Integration` class are
    properly populated.

    Parameters
    ----------
    client : Client
        Client to use for interacting with the Nextmv Cloud API.
    integration_id : str
        The unique identifier of the integration to retrieve.

    Returns
    -------
    Integration
        The retrieved integration instance.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> from nextmv.cloud import Client, Integration
    >>> client = Client(api_key="your_api_key")
    >>> integration = Integration.get(client=client, integration_id="your_integration_id")
    >>> print(integration.to_dict())
    """

    response = client.request(
        method="GET",
        endpoint=f"v1/integrations/{integration_id}",
    )
    response_dict = response.json()
    response_dict["client"] = client

    return cls.from_dict(response_dict)

integration_id class-attribute instance-attribute

integration_id: str = Field(
    serialization_alias="id",
    validation_alias=AliasChoices("id", "integration_id"),
)

The unique identifier of the integration.

integration_type class-attribute instance-attribute

integration_type: IntegrationType | None = Field(
    serialization_alias="type",
    validation_alias=AliasChoices(
        "type", "integration_type"
    ),
    default=None,
)

The type of the integration (runtime or data).

is_global class-attribute instance-attribute

is_global: bool = Field(
    serialization_alias="global",
    validation_alias=AliasChoices("global", "is_global"),
    default=False,
)

Indicates whether the integration is global (available to all applications in the account).

model_post_init

model_post_init(__context) -> None

Validations done after model initialization.

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

def model_post_init(self, __context) -> None:
    """
    Validations done after model initialization.
    """

    self.endpoint = self.endpoint.format(id=self.integration_id)

name class-attribute instance-attribute

name: str | None = None

The name of the integration.

new classmethod

new(
    client: Client,
    name: str,
    integration_type: IntegrationType | str,
    exec_types: list[ManifestType | str],
    provider: IntegrationProvider | str,
    provider_config: dict[str, Any],
    integration_id: str | None = None,
    description: str | None = None,
    is_global: bool = False,
    application_ids: list[str] | None = None,
    exist_ok: bool = False,
) -> Integration

Create a new integration directly in Nextmv Cloud.

PARAMETER	DESCRIPTION
client	Client to use for interacting with the Nextmv Cloud API. TYPE: Client
name	The name of the integration. TYPE: str
integration_type	The type of the integration. Please refer to the IntegrationType enum for possible values. TYPE: IntegrationType | str
exec_types	List of execution types supported by the integration. Please refer to the ManifestType enum for possible values. TYPE: list[ManifestType | str]
provider	The provider of the integration. Please refer to the IntegrationProvider enum for possible values. TYPE: IntegrationProvider | str
provider_config	Configuration specific to the integration provider. TYPE: dict[str, Any]
integration_id	The unique identifier of the integration. If not provided, it will be generated automatically. TYPE: str DEFAULT: None
description	An optional description of the integration. TYPE: str DEFAULT: None
is_global	Indicates whether the integration is global (available to all applications in the account). Default is False. TYPE: bool DEFAULT: False
application_ids	List of application IDs that have access to this integration. TYPE: list[str] DEFAULT: None
exist_ok	If True and an integration with the same ID already exists, return the existing integration instead of creating a new one. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
Integration	The created integration instance.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.
ValueError	If both is_global is True and application_ids is provided.

Examples:

>>> from nextmv.cloud import Client, Integration, IntegrationType, IntegrationProvider, ManifestType
>>> client = Client(api_key="your_api_key")
>>> integration = Integration.new(
...     client=client,
...     name="my_integration",
...     integration_type=IntegrationType.RUNTIME,
...     exec_types=[ManifestType.PYTHON],
...     provider=IntegrationProvider.DBX,
...     provider_config={"config_key": "config_value"},
... )
>>> print(integration.to_dict())

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

@classmethod
def new(  # noqa: C901
    cls,
    client: Client,
    name: str,
    integration_type: IntegrationType | str,
    exec_types: list[ManifestType | str],
    provider: IntegrationProvider | str,
    provider_config: dict[str, Any],
    integration_id: str | None = None,
    description: str | None = None,
    is_global: bool = False,
    application_ids: list[str] | None = None,
    exist_ok: bool = False,
) -> "Integration":
    """
    Create a new integration directly in Nextmv Cloud.

    Parameters
    ----------
    client : Client
        Client to use for interacting with the Nextmv Cloud API.
    name : str
        The name of the integration.
    integration_type : IntegrationType | str
        The type of the integration. Please refer to the `IntegrationType`
        enum for possible values.
    exec_types : list[ManifestType | str]
        List of execution types supported by the integration. Please refer
        to the `ManifestType` enum for possible values.
    provider : IntegrationProvider | str
        The provider of the integration. Please refer to the
        `IntegrationProvider` enum for possible values.
    provider_config : dict[str, Any]
        Configuration specific to the integration provider.
    integration_id : str, optional
        The unique identifier of the integration. If not provided,
        it will be generated automatically.
    description : str, optional
        An optional description of the integration.
    is_global : bool, optional, default=False
        Indicates whether the integration is global (available to all
        applications in the account). Default is False.
    application_ids : list[str], optional
        List of application IDs that have access to this integration.
    exist_ok : bool, default=False
        If True and an integration with the same ID already exists,
        return the existing integration instead of creating a new one.

    Returns
    -------
    Integration
        The created integration instance.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.
    ValueError
        If both `is_global` is True and `application_ids` is provided.

    Examples
    --------
    >>> from nextmv.cloud import Client, Integration, IntegrationType, IntegrationProvider, ManifestType
    >>> client = Client(api_key="your_api_key")
    >>> integration = Integration.new(
    ...     client=client,
    ...     name="my_integration",
    ...     integration_type=IntegrationType.RUNTIME,
    ...     exec_types=[ManifestType.PYTHON],
    ...     provider=IntegrationProvider.DBX,
    ...     provider_config={"config_key": "config_value"},
    ... )
    >>> print(integration.to_dict())
    """

    if is_global and application_ids is not None:
        raise ValueError("An integration cannot be global and have specific application IDs.")
    elif not is_global and application_ids is None:
        raise ValueError("A non-global integration must have specific application IDs.")

    if integration_id is None:
        integration_id = safe_id("integration")

    if exist_ok:
        try:
            integration = cls.get(client=client, integration_id=integration_id)
            return integration
        except Exception:
            pass

    if not isinstance(integration_type, IntegrationType):
        integration_type = IntegrationType(integration_type)

    if not all(isinstance(exec_type, ManifestType) for exec_type in exec_types):
        exec_types = [ManifestType(exec_type) for exec_type in exec_types]

    if not isinstance(provider, IntegrationProvider):
        provider = IntegrationProvider(provider)

    payload = {
        "id": integration_id,
        "name": name,
        "global": is_global,
        "type": integration_type.value,
        "exec_types": [exec_type.value for exec_type in exec_types],
        "provider": provider.value,
        "provider_config": provider_config,
    }

    if description is not None:
        payload["description"] = description

    if application_ids is not None:
        payload["application_ids"] = application_ids

    response = client.request(
        method="POST",
        endpoint="v1/integrations",
        payload=payload,
    )
    response_dict = response.json()
    response_dict["client"] = client
    integration = cls.from_dict(response_dict)

    return integration

provider class-attribute instance-attribute

provider: IntegrationProvider | None = None

The provider of the integration.

provider_config class-attribute instance-attribute

provider_config: dict[str, Any] | None = None

Configuration specific to the integration provider.

update

update(
    name: str | None = None,
    integration_type: IntegrationType | str | None = None,
    exec_types: list[ManifestType | str] | None = None,
    provider: IntegrationProvider | str | None = None,
    provider_config: dict[str, Any] | None = None,
    description: str | None = None,
    is_global: bool | None = None,
    application_ids: list[str] | None = None,
) -> Integration

Updates the integration in Nextmv Cloud.

PARAMETER	DESCRIPTION
name	The new name of the integration. TYPE: str DEFAULT: None
integration_type	The new type of the integration. Please refer to the IntegrationType enum for possible values. TYPE: IntegrationType | str DEFAULT: None
exec_types	New list of execution types supported by the integration. Please refer to the ManifestType enum for possible values. TYPE: list[ManifestType | str] DEFAULT: None
provider	The new provider of the integration. Please refer to the IntegrationProvider enum for possible values. TYPE: IntegrationProvider | str DEFAULT: None
provider_config	New configuration specific to the integration provider. TYPE: dict[str, Any] DEFAULT: None
description	The new description of the integration. TYPE: str DEFAULT: None
is_global	Indicates whether the integration is global (available to all applications in the account). If not provided, the current value is preserved. TYPE: bool DEFAULT: None
application_ids	New list of application IDs that have access to this integration. TYPE: list[str] DEFAULT: None

RETURNS	DESCRIPTION
Integration	The updated integration instance.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> from nextmv.cloud import Client, Integration
>>> client = Client(api_key="your_api_key")
>>> integration = Integration.get(client=client, integration_id="your_integration_id")
>>> updated_integration = integration.update(name="new_name")
>>> print(updated_integration.to_dict())

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

def update(  # noqa: C901
    self,
    name: str | None = None,
    integration_type: IntegrationType | str | None = None,
    exec_types: list[ManifestType | str] | None = None,
    provider: IntegrationProvider | str | None = None,
    provider_config: dict[str, Any] | None = None,
    description: str | None = None,
    is_global: bool | None = None,
    application_ids: list[str] | None = None,
) -> "Integration":
    """
    Updates the integration in Nextmv Cloud.

    Parameters
    ----------
    name : str, optional
        The new name of the integration.
    integration_type : IntegrationType | str, optional
        The new type of the integration. Please refer to the `IntegrationType`
        enum for possible values.
    exec_types : list[ManifestType | str], optional
        New list of execution types supported by the integration. Please refer
        to the `ManifestType` enum for possible values.
    provider : IntegrationProvider | str, optional
        The new provider of the integration. Please refer to the
        `IntegrationProvider` enum for possible values.
    provider_config : dict[str, Any], optional
        New configuration specific to the integration provider.
    description : str, optional
        The new description of the integration.
    is_global : bool, optional
        Indicates whether the integration is global (available to all
        applications in the account). If not provided, the current value
        is preserved.
    application_ids : list[str], optional
        New list of application IDs that have access to this integration.

    Returns
    -------
    Integration
        The updated integration instance.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> from nextmv.cloud import Client, Integration
    >>> client = Client(api_key="your_api_key")
    >>> integration = Integration.get(client=client, integration_id="your_integration_id")
    >>> updated_integration = integration.update(name="new_name")
    >>> print(updated_integration.to_dict())
    """

    integration = self.get(client=self.client, integration_id=self.integration_id)
    integration_dict = integration.to_dict()
    payload = integration_dict

    if name is not None:
        payload["name"] = name

    if integration_type is not None:
        if not isinstance(integration_type, IntegrationType):
            integration_type = IntegrationType(integration_type)
        payload["type"] = integration_type.value

    if exec_types is not None:
        if not all(isinstance(exec_type, ManifestType) for exec_type in exec_types):
            exec_types = [ManifestType(exec_type) for exec_type in exec_types]
        payload["exec_types"] = [exec_type.value for exec_type in exec_types]

    if provider is not None:
        if not isinstance(provider, IntegrationProvider):
            provider = IntegrationProvider(provider)
        payload["provider"] = provider.value

    if provider_config is not None:
        payload["provider_config"] = provider_config

    if description is not None:
        payload["description"] = description

    if is_global is not None:
        payload["global"] = is_global

    if application_ids is not None:
        payload["application_ids"] = application_ids

    # Final validation: ensure invariants are met.
    if payload["global"] is True and payload.get("application_ids"):
        raise ValueError(
            "An integration cannot be global and have application_ids. "
            "To make an integration global, call update(is_global=True, application_ids=[])."
        )
    if payload["global"] is False and not payload.get("application_ids"):
        raise ValueError(
            "A non-global integration must have specific application IDs. "
            "Provide application_ids with at least one ID, or set is_global=True."
        )

    response = self.client.request(
        method="PUT",
        endpoint=self.endpoint,
        payload=payload,
    )
    response_dict = response.json()
    response_dict["client"] = self.client
    integration = self.from_dict(response_dict)

    return integration

updated_at class-attribute instance-attribute

updated_at: datetime | None = None

The timestamp when the integration was last updated.

IntegrationProvider

Bases: str, Enum

The provider of an integration.

You can import the IntegrationProvider class directly from cloud:

from nextmv.cloud import IntegrationProvider

ATTRIBUTE	DESCRIPTION
DBX	Indicates a Databricks integration. TYPE: str
UNKNOWN	Indicates an unknown integration provider. TYPE: str

DBX class-attribute instance-attribute

DBX = 'dbx'

Indicates a Databricks integration.

UNKNOWN class-attribute instance-attribute

UNKNOWN = 'unknown'

Indicates an unknown integration provider.

IntegrationType

Bases: str, Enum

The type of an integration.

You can import the IntegrationType class directly from cloud:

from nextmv.cloud import IntegrationType

ATTRIBUTE	DESCRIPTION
RUNTIME	Indicates a runtime integration. TYPE: str
DATA	Indicates a data integration. TYPE: str

DATA class-attribute instance-attribute

DATA = 'data'

Indicates a data integration.

RUNTIME class-attribute instance-attribute

RUNTIME = 'runtime'

Indicates a runtime integration.

list_integrations

list_integrations(client: Client) -> list[Integration]

List all integrations in Nextmv Cloud for the given client.

You can import the list_integrations method directly from cloud:

from nextmv.cloud import list_integrations

PARAMETER	DESCRIPTION
client	Client to use for interacting with the Nextmv Cloud API. TYPE: Client

RETURNS	DESCRIPTION
list[Integration]	List of integrations.

RAISES	DESCRIPTION
HTTPError	If the response status code is not 2xx.

Examples:

>>> from nextmv.cloud import Client, list_integrations
>>> client = Client(api_key="your_api_key")
>>> integrations = list_integrations(client=client)
>>> for integration in integrations:
...     print(integration.to_dict())

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

def list_integrations(client: Client) -> list[Integration]:
    """
    List all integrations in Nextmv Cloud for the given client.

    You can import the `list_integrations` method directly from `cloud`:

    ```python
    from nextmv.cloud import list_integrations
    ```

    Parameters
    ----------
    client : Client
        Client to use for interacting with the Nextmv Cloud API.

    Returns
    -------
    list[Integration]
        List of integrations.

    Raises
    ------
    requests.HTTPError
        If the response status code is not 2xx.

    Examples
    --------
    >>> from nextmv.cloud import Client, list_integrations
    >>> client = Client(api_key="your_api_key")
    >>> integrations = list_integrations(client=client)
    >>> for integration in integrations:
    ...     print(integration.to_dict())
    """

    response = client.request(
        method="GET",
        endpoint="v1/integrations",
    )
    response_dict = response.json()
    integrations = []
    for integration_data in response_dict.get("items", []):
        integration_data["client"] = client
        integration = Integration.from_dict(integration_data)
        integrations.append(integration)

    return integrations