<filename>dialogflow-cx/delegators/agent_delegator.py
# Copyright 2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Agent Delegator module. Coordinates agent state with Dialogflow."""


import dialogflow_sample as ds
import google.api_core.exceptions
import google.auth
import google.cloud.dialogflowcx as cx

from .client_delegator import ClientDelegator


class AgentDelegator(ClientDelegator):
    """Class for organizing interactions with the Dialogflow Agent API."""

    _DEFAULT_LANGUAGE_CODE = "en"
    _DEFAULT_TIME_ZONE = "America/Los_Angeles"
    _CLIENT_CLASS = cx.AgentsClient

    def __init__(self, controller: ds.DialogflowSample, **kwargs) -> None:
        super().__init__(controller, **kwargs)
        self._agent = None
        self.default_language_code = kwargs.get(
            "default_language_code", self._DEFAULT_LANGUAGE_CODE
        )
        self.time_zone = kwargs.get("time_zone", self._DEFAULT_TIME_ZONE)

    def setup(self):
        """Initializes the agent delegator."""
        try:
            agent = cx.Agent(
                display_name=self.display_name,
                default_language_code=self.default_language_code,
                time_zone=self.time_zone,
            )
            request = {"agent": agent, "parent": self.parent}
            self._agent = self.client.create_agent(request=request)
        except google.api_core.exceptions.AlreadyExists:
            request = cx.ListAgentsRequest(
                parent=self.parent,
            )
            for agent in self.client.list_agents(request=request):
                if agent.display_name == self.display_name:
                    request = cx.GetAgentRequest(
                        name=agent.name,
                    )
                    self._agent = self.client.get_agent(request=request)
                    break

    def tear_down(self):
        """Destroys the Dialogflow agent."""
        request = cx.DeleteAgentRequest(name=self.agent.name)
        try:
            self.client.delete_agent(request=request)
            self._agent = None
        except google.api_core.exceptions.NotFound:
            pass

    @property
    def parent(self):
        """Accesses the parent of the agent."""
        return f"projects/{self.controller.project_id}/locations/{self.controller.location}"

    @property
    def agent(self):
        """Agent set in Dialogflow."""
        if not self._agent:
            raise RuntimeError("Agent not yet created")
        return self._agent

    @property
    def start_flow(self):
        """Accesses the start flow of the agent."""
        return self.agent.start_flow

<filename>dialogflow-cx/delegators/agent_delegator.py
# Copyright 2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Agent Delegator module. Coordinates agent state with Dialogflow."""


import dialogflow_sample as ds
import google.api_core.exceptions
import google.auth
import google.cloud.dialogflowcx as cx

from .client_delegator import ClientDelegator


class AgentDelegator(ClientDelegator):
    """Class for organizing interactions with the Dialogflow Agent API."""

    _DEFAULT_LANGUAGE_CODE = "en"
    _DEFAULT_TIME_ZONE = "America/Los_Angeles"
    _CLIENT_CLASS = cx.AgentsClient

    def __init__(self, controller: ds.DialogflowSample, **kwargs) -> None:
        super().__init__(controller, **kwargs)
        self._agent = None
        self.default_language_code = kwargs.get(
            "default_language_code", self._DEFAULT_LANGUAGE_CODE
        )
        self.time_zone = kwargs.get("time_zone", self._DEFAULT_TIME_ZONE)

    def setup(self):
        """Initializes the agent delegator."""
        try:
            agent = cx.Agent(
                display_name=self.display_name,
                default_language_code=self.default_language_code,
                time_zone=self.time_zone,
            )
            request = {"agent": agent, "parent": self.parent}
            self._agent = self.client.create_agent(request=request)
        except google.api_core.exceptions.AlreadyExists:
            request = cx.ListAgentsRequest(
                parent=self.parent,
            )
            for agent in self.client.list_agents(request=request):
                if agent.display_name == self.display_name:
                    request = cx.GetAgentRequest(
                        name=agent.name,
                    )
                    self._agent = self.client.get_agent(request=request)
                    break

    def tear_down(self):
        """Destroys the Dialogflow agent."""
        request = cx.DeleteAgentRequest(name=self.agent.name)
        try:
            self.client.delete_agent(request=request)
            self._agent = None
        except google.api_core.exceptions.NotFound:
            pass

    @property
    def parent(self):
        """Accesses the parent of the agent."""
        return f"projects/{self.controller.project_id}/locations/{self.controller.location}"

    @property
    def agent(self):
        """Agent set in Dialogflow."""
        if not self._agent:
            raise RuntimeError("Agent not yet created")
        return self._agent

    @property
    def start_flow(self):
        """Accesses the start flow of the agent."""
        return self.agent.start_flow

# Copyright 2022 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # #      http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. Agent Delegator module. Coordinates agent state with Dialogflow. Class for organizing interactions with the Dialogflow Agent API. Initializes the agent delegator. Destroys the Dialogflow agent. Accesses the parent of the agent. Agent set in Dialogflow. Accesses the start flow of the agent.