Source code for airflow.providers.zendesk.hooks.zendesk

#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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.
from __future__ import annotations

from typing import TYPE_CHECKING, Any

from zenpy import Zenpy

from airflow.hooks.base import BaseHook

if TYPE_CHECKING:
    from zenpy.lib.api import BaseApi
    from zenpy.lib.api_objects import JobStatus, Ticket, TicketAudit
    from zenpy.lib.generator import SearchResultGenerator



[docs]
class ZendeskHook(BaseHook):
    """
    Interact with Zendesk. This hook uses the Zendesk conn_id.

    :param zendesk_conn_id: The Airflow connection used for Zendesk credentials.
    """


[docs]
    conn_name_attr = "zendesk_conn_id"


[docs]
    default_conn_name = "zendesk_default"


[docs]
    conn_type = "zendesk"


[docs]
    hook_name = "Zendesk"


    @classmethod

[docs]
    def get_ui_field_behaviour(cls) -> dict[str, Any]:
        return {
            "hidden_fields": ["schema", "port", "extra"],
            "relabeling": {"host": "Zendesk domain", "login": "Zendesk email"},
        }


    def __init__(self, zendesk_conn_id: str = default_conn_name) -> None:
        super().__init__()

[docs]
        self.zendesk_conn_id = zendesk_conn_id


[docs]
        self.base_api: BaseApi | None = None

        zenpy_client, url = self._init_conn()

[docs]
        self.zenpy_client = zenpy_client

        self.__url = url

[docs]
        self.get = self.zenpy_client.users._get


    def _init_conn(self) -> tuple[Zenpy, str]:
        """
        Create the Zenpy Client for our Zendesk connection.

        :return: zenpy.Zenpy client and the url for the API.
        """
        conn = self.get_connection(self.zendesk_conn_id)
        url = "https://" + conn.host
        domain = conn.host
        subdomain: str | None = None
        if conn.host.count(".") >= 2:
            dot_splitted_string = conn.host.rsplit(".", 2)
            subdomain = dot_splitted_string[0]
            domain = ".".join(dot_splitted_string[1:])
        return Zenpy(domain=domain, subdomain=subdomain, email=conn.login, password=conn.password), url


[docs]
    def get_conn(self) -> Zenpy:
        """
        Get the underlying Zenpy client.

        :return: zenpy.Zenpy client.
        """
        return self.zenpy_client



[docs]
    def get_ticket(self, ticket_id: int) -> Ticket:
        """
        Retrieve ticket.

        :return: Ticket object retrieved.
        """
        return self.zenpy_client.tickets(id=ticket_id)



[docs]
    def search_tickets(self, **kwargs) -> SearchResultGenerator:
        """
        Search tickets.

        :param kwargs: (optional) Search fields given to the zenpy search method.
        :return: SearchResultGenerator of Ticket objects.
        """
        return self.zenpy_client.search(type="ticket", **kwargs)



[docs]
    def create_tickets(self, tickets: Ticket | list[Ticket], **kwargs) -> TicketAudit | JobStatus:
        """
        Create tickets.

        :param tickets: Ticket or List of Ticket to create.
        :param kwargs: (optional) Additional fields given to the zenpy create method.
        :return: A TicketAudit object containing information about the Ticket created.
            When sending bulk request, returns a JobStatus object.
        """
        return self.zenpy_client.tickets.create(tickets, **kwargs)



[docs]
    def update_tickets(self, tickets: Ticket | list[Ticket], **kwargs) -> TicketAudit | JobStatus:
        """
        Update tickets.

        :param tickets: Updated Ticket or List of Ticket object to update.
        :param kwargs: (optional) Additional fields given to the zenpy update method.
        :return: A TicketAudit object containing information about the Ticket updated.
            When sending bulk request, returns a JobStatus object.
        """
        return self.zenpy_client.tickets.update(tickets, **kwargs)



[docs]
    def delete_tickets(self, tickets: Ticket | list[Ticket], **kwargs) -> None:
        """
        Delete tickets, returns nothing on success and raises APIException on failure.

        :param tickets: Ticket or List of Ticket to delete.
        :param kwargs: (optional) Additional fields given to the zenpy delete method.
        :return:
        """
        return self.zenpy_client.tickets.delete(tickets, **kwargs)