Deployments¶

def list_deployments(self) -> list[types.Deployment]:
    """Lists deployments within a specific app."""
    request = types.ListDeploymentsRequest(parent=self.app_name)
    response = self.client.list_deployments(request=request)
    return list(response)

def get_deployments_map(self, reverse: bool = False) -> dict[str, str]:
    """Creates a map of Deployment full names to display names.

    Args:
        reverse: If True, map display_name -> name.
    """
    deployments = self.list_deployments()
    deployments_dict: dict[str, str] = {}

    for deployment in deployments:
        display_name = deployment.display_name
        name = deployment.name
        if display_name and name:
            if reverse:
                deployments_dict[display_name] = name
            else:
                deployments_dict[name] = display_name
    return deployments_dict

def get_deployment(self, deployment_id: str) -> types.Deployment:
    """Gets a specific deployment."""
    request = types.GetDeploymentRequest(
        name=f"{self.app_name}/deployments/{deployment_id}"
    )
    return self.client.get_deployment(request=request)

def create_deployment(
    self,
    deployment_id: str,
    display_name: str,
    app_version: str,
    channel_type: ChannelType | str = ChannelType.API,
    modality: Modality | str | None = None,
    theme: Theme | str | None = None,
    web_widget_title: str | None = None,
    disable_dtmf: bool = False,
    disable_barge_in_control: bool = False,
    persona_property: Persona
    | types.ChannelProfile.PersonaProperty
    | str
    | None = None,
    noise_suppression_level: str | None = None,
    traffic_split: dict[str, int] | None = None,
) -> types.Deployment:
    """Creates a new deployment with specified configuration.

    Args:
        deployment_id: The ID to use for the deployment.
        display_name: The display name of the deployment.
        app_version: App version name or ID to deploy.
        channel_type: The channel type (e.g. ChannelType.API).
        modality: Web widget modality (WEB_UI only).
        theme: Web widget theme (WEB_UI only).
        web_widget_title: Web widget title (WEB_UI only).
        disable_dtmf: Whether DTMF is disabled.
        disable_barge_in_control: Whether barge-in control is disabled.
        persona_property: Persona property (CONCISE, CHATTY).
            Accepts Deployments.Persona, string, or
            types.ChannelProfile.PersonaProperty.
        noise_suppression_level: Noise suppression level (low, moderate,
            high, very_high).
        traffic_split: Traffic split configuration between versions.

    Note: `modality`, `theme`, and `web_widget_title` are only applicable
    when `channel_type` is `ChannelType.WEB_UI`.
    """

    if app_version and not app_version.startswith("projects/"):
        app_version = f"{self.app_name}/versions/{app_version}"

    deployment = types.Deployment(
        display_name=display_name, app_version=app_version
    )

    # Convert string to enum if needed
    if isinstance(channel_type, str):
        channel_type = self.ChannelType[channel_type.upper()]

    channel_profile = types.ChannelProfile()

    channel_profile.channel_type = getattr(
        types.common.ChannelProfile.ChannelType, channel_type.value
    )

    channel_profile.disable_dtmf = disable_dtmf
    channel_profile.disable_barge_in_control = disable_barge_in_control

    if persona_property:
        channel_profile.persona_property = self._build_persona_property(
            persona_property
        )

    if noise_suppression_level:
        channel_profile.noise_suppression_level = (
            self._validate_noise_suppression_level(noise_suppression_level)
        )

    if channel_type == self.ChannelType.WEB_UI:
        wwc_kwargs = {
            "modality": modality or self.Modality.CHAT_AND_VOICE,
            "theme": theme or self.Theme.LIGHT,
        }
        if web_widget_title:
            wwc_kwargs["web_widget_title"] = web_widget_title

        wwc = self._build_web_widget_config(wwc_kwargs)
        if wwc:
            channel_profile.web_widget_config = wwc

    deployment.channel_profile = channel_profile

    if traffic_split:
        if len(traffic_split) < 2:
            raise ValueError(
                "Traffic split requires at least two versions."
            )
        if hasattr(types, "ExperimentConfig"):
            versions_client = Versions(
                app_name=self.app_name, creds=self.creds
            )
            existing_versions = versions_client.list_versions()
            existing_version_names = [v.name for v in existing_versions]

            experiment_config = types.ExperimentConfig()
            version_release = types.ExperimentConfig.VersionRelease()
            version_release.state = types.ExperimentConfig.State.RUNNING
            for version, split in traffic_split.items():
                v_name = version
                if not v_name.startswith("projects/"):
                    v_name = f"{self.app_name}/versions/{version}"

                if v_name not in existing_version_names:
                    raise ValueError(
                        f"Version {v_name} does not exist. Valid versions: "
                        f"{[v.split('/')[-1] for v in existing_version_names]}"  # noqa: E501
                    )

                allocation = types.ExperimentConfig.VersionRelease.TrafficAllocation()  # noqa: E501
                allocation.app_version = v_name
                allocation.traffic_percentage = split
                version_release.traffic_allocations.append(allocation)

            experiment_config.version_release = version_release
            deployment.experiment_config = experiment_config
        else:
            raise NotImplementedError(
                "traffic_split requires ExperimentConfig which is "
                "not available in the current API schema."
            )

    request = types.CreateDeploymentRequest(
        parent=self.app_name,
        deployment_id=deployment_id,
        deployment=deployment,
    )
    return self.client.create_deployment(request=request)

def update_deployment(
    self, deployment_id: str, **kwargs: typing.Any
) -> types.Deployment:
    """Updates specific fields of an existing Deployment.

    Args:
        deployment_id: The ID of the deployment to update.
        **kwargs: Fields to update. Supported fields include:
            - display_name: New display name.
            - app_version: New app version name or ID.
            - channel_type: New ChannelType or string.
            - modality: Web widget modality (WEB_UI only).
            - theme: Web widget theme (WEB_UI only).
            - web_widget_title: Web widget title (WEB_UI only).
            - disable_dtmf: Whether DTMF is disabled.
            - disable_barge_in_control: Whether barge-in control is
                disabled.
            - persona_property: Persona property (CONCISE, CHATTY).
                Accepts Deployments.Persona, string, or
                types.ChannelProfile.PersonaProperty.
            - noise_suppression_level: Noise suppression level (low,
                moderate, high, very_high).
            - traffic_split: Traffic split configuration between versions.
    """
    deployment = types.Deployment(
        name=f"{self.app_name}/deployments/{deployment_id}"
    )
    mask_paths = []

    channel_profile_fields = [
        "channel_type",
        "modality",
        "theme",
        "web_widget_title",
        "disable_dtmf",
        "disable_barge_in_control",
        "persona_property",
        "noise_suppression_level",
    ]

    has_channel_profile_update = any(
        k in kwargs for k in channel_profile_fields
    )

    if has_channel_profile_update:
        channel_profile = types.ChannelProfile()

        if "channel_type" in kwargs:
            channel_type = kwargs.pop("channel_type")
            if isinstance(channel_type, str):
                channel_type = self.ChannelType[channel_type.upper()]
            channel_profile.channel_type = getattr(
                types.common.ChannelProfile.ChannelType, channel_type.value
            )
            mask_paths.append("channel_profile.channel_type")

        if "disable_dtmf" in kwargs:
            channel_profile.disable_dtmf = kwargs.pop("disable_dtmf")
            mask_paths.append("channel_profile.disable_dtmf")

        if "disable_barge_in_control" in kwargs:
            channel_profile.disable_barge_in_control = kwargs.pop(
                "disable_barge_in_control"
            )
            mask_paths.append("channel_profile.disable_barge_in_control")

        if "persona_property" in kwargs:
            persona_prop = kwargs.pop("persona_property")
            if persona_prop:
                channel_profile.persona_property = (
                    self._build_persona_property(persona_prop)
                )
            mask_paths.append("channel_profile.persona_property")

        if "noise_suppression_level" in kwargs:
            noise_level = kwargs.pop("noise_suppression_level")
            if noise_level:
                channel_profile.noise_suppression_level = (
                    self._validate_noise_suppression_level(noise_level)
                )
            mask_paths.append("channel_profile.noise_suppression_level")

        wwc = self._build_web_widget_config(kwargs, mask_paths)
        if wwc:
            channel_profile.web_widget_config = wwc

        deployment.channel_profile = channel_profile

    if "traffic_split" in kwargs:
        traffic_split = kwargs.pop("traffic_split")
        if len(traffic_split) < 2:
            raise ValueError(
                "Traffic split requires at least two versions."
            )
        if hasattr(types, "ExperimentConfig"):
            versions_client = Versions(
                app_name=self.app_name, creds=self.creds
            )
            existing_versions = versions_client.list_versions()
            existing_version_names = [v.name for v in existing_versions]

            experiment_config = types.ExperimentConfig()
            version_release = types.ExperimentConfig.VersionRelease()
            version_release.state = types.ExperimentConfig.State.RUNNING
            for version, split in traffic_split.items():
                v_name = version
                if not v_name.startswith("projects/"):
                    v_name = f"{self.app_name}/versions/{version}"

                if v_name not in existing_version_names:
                    raise ValueError(
                        f"Version {v_name} does not exist. Valid versions: "
                        f"{[v.split('/')[-1] for v in existing_version_names]}"  # noqa: E501
                    )

                allocation = types.ExperimentConfig.VersionRelease.TrafficAllocation()  # noqa: E501
                allocation.app_version = v_name
                allocation.traffic_percentage = split
                version_release.traffic_allocations.append(allocation)

            experiment_config.version_release = version_release
            deployment.experiment_config = experiment_config
            mask_paths.append("experiment_config")
        else:
            raise NotImplementedError(
                "traffic_split requires ExperimentConfig which is "
                "not available in the current API schema."
            )
    elif "app_version" in kwargs:
        # If promoting a new version without a traffic split,
        # clear any existing experiment
        deployment.experiment_config = types.ExperimentConfig()
        mask_paths.append("experiment_config")

    # Handle remaining kwargs as top-level fields
    for key, value in kwargs.items():
        val_to_set = value
        is_app_ver = key == "app_version"
        if is_app_ver and value and not value.startswith("projects/"):
            val_to_set = f"{self.app_name}/versions/{value}"
        setattr(deployment, key, val_to_set)
        mask_paths.append(key)

    request = types.UpdateDeploymentRequest(
        deployment=deployment,
        update_mask=field_mask_pb2.FieldMask(paths=mask_paths),
    )
    return self.client.update_deployment(request=request)

def delete_deployment(self, deployment_id: str) -> None:
    """Deletes a specific deployment."""
    request = types.DeleteDeploymentRequest(
        name=f"{self.app_name}/deployments/{deployment_id}"
    )
    self.client.delete_deployment(request=request)