deepgram.clients.selfhosted.v1.async_client

  1# Copyright 2023-2024 Deepgram SDK contributors. All Rights Reserved.
  2# Use of this source code is governed by a MIT license that can be found in the LICENSE file.
  3# SPDX-License-Identifier: MIT
  4
  5import logging
  6from typing import Optional
  7
  8import httpx
  9
 10from ....utils import verboselogs
 11from ....options import DeepgramClientOptions
 12from ...common import AbstractAsyncRestClient
 13
 14
 15class AsyncSelfHostedClient(AbstractAsyncRestClient):
 16    """
 17    Client for interacting with Deepgram's on-premises API.
 18
 19    This class provides methods to manage and interact with on-premises projects and distribution credentials.
 20
 21    Args:
 22        config (DeepgramClientOptions): all the options for the client.
 23    """
 24
 25    _logger: verboselogs.VerboseLogger
 26    _config: DeepgramClientOptions
 27    _endpoint: str
 28
 29    def __init__(self, config: DeepgramClientOptions):
 30        self._logger = verboselogs.VerboseLogger(__name__)
 31        self._logger.addHandler(logging.StreamHandler())
 32        self._logger.setLevel(config.verbose)
 33        self._config = config
 34        self._endpoint = "v1/projects"
 35        super().__init__(config)
 36
 37    async def list_onprem_credentials(
 38        self,
 39        project_id: str,
 40        timeout: Optional[httpx.Timeout] = None,
 41        **kwargs,
 42    ):
 43        """
 44        List all on-premises distribution credentials for a project.
 45        """
 46        return self.list_selfhosted_credentials(project_id, timeout=timeout, **kwargs)
 47
 48    async def list_selfhosted_credentials(
 49        self,
 50        project_id: str,
 51        timeout: Optional[httpx.Timeout] = None,
 52        **kwargs,
 53    ):
 54        """
 55        List all on-premises distribution credentials for a project.
 56        """
 57        self._logger.debug("SelfHostedClient.list_selfhosted_credentials ENTER")
 58        url = f"{self._config.url}/{self._endpoint}/{project_id}/selfhosted/distribution/credentials"
 59        self._logger.info("url: %s", url)
 60        self._logger.info("project_id: %s", project_id)
 61        res = await self.get(url, timeout=timeout, **kwargs)
 62        self._logger.verbose("result: %s", res)
 63        self._logger.notice("list_selfhosted_credentials succeeded")
 64        self._logger.debug("SelfHostedClient.list_selfhosted_credentials LEAVE")
 65        return res
 66
 67    async def get_onprem_credentials(
 68        self,
 69        project_id: str,
 70        distribution_credentials_id: str,
 71        timeout: Optional[httpx.Timeout] = None,
 72        **kwargs,
 73    ):
 74        """
 75        Get a specific on-premises distribution credential for a project.
 76        """
 77        return self.get_selfhosted_credentials(
 78            project_id, distribution_credentials_id, timeout=timeout, **kwargs
 79        )
 80
 81    async def get_selfhosted_credentials(
 82        self,
 83        project_id: str,
 84        distribution_credentials_id: str,
 85        timeout: Optional[httpx.Timeout] = None,
 86        **kwargs,
 87    ):
 88        """
 89        Get a specific on-premises distribution credential for a project.
 90        """
 91        self._logger.debug("SelfHostedClient.get_selfhosted_credentials ENTER")
 92        url = f"{self._config.url}/{self._endpoint}/{project_id}/selfhosted/distribution/credentials/{distribution_credentials_id}"
 93        self._logger.info("url: %s", url)
 94        self._logger.info("project_id: %s", project_id)
 95        self._logger.info(
 96            "distribution_credentials_id: %s", distribution_credentials_id
 97        )
 98        res = await self.get(url, timeout=timeout, **kwargs)
 99        self._logger.verbose("result: %s", res)
100        self._logger.notice("get_selfhosted_credentials succeeded")
101        self._logger.debug("SelfHostedClient.get_selfhosted_credentials LEAVE")
102        return res
103
104    async def create_onprem_credentials(
105        self,
106        project_id: str,
107        options,
108        timeout: Optional[httpx.Timeout] = None,
109        **kwargs,
110    ):
111        """
112        Create a new on-premises distribution credential for a project.
113        """
114        return self.create_onprem_credentials(project_id, options, timeout, **kwargs)
115
116    async def create_selfhosted_credentials(
117        self,
118        project_id: str,
119        options,
120        timeout: Optional[httpx.Timeout] = None,
121        **kwargs,
122    ):
123        """
124        Create a new on-premises distribution credential for a project.
125        """
126        self._logger.debug("SelfHostedClient.create_selfhosted_credentials ENTER")
127        url = f"{self._config.url}/{self._endpoint}/{project_id}/selfhosted/distribution/credentials/"
128        self._logger.info("url: %s", url)
129        self._logger.info("project_id: %s", project_id)
130        self._logger.info("options: %s", options)
131        res = await self.post(url, json=options, timeout=timeout, **kwargs)
132        self._logger.verbose("result: %s", res)
133        self._logger.notice("create_selfhosted_credentials succeeded")
134        self._logger.debug("SelfHostedClient.create_selfhosted_credentials LEAVE")
135        return res
136
137    async def delete_onprem_credentials(
138        self,
139        project_id: str,
140        distribution_credentials_id: str,
141        timeout: Optional[httpx.Timeout] = None,
142        **kwargs,
143    ):
144        """
145        Delete an on-premises distribution credential for a project.
146        """
147        return self.delete_selfhosted_credentials(
148            project_id, distribution_credentials_id, timeout=timeout, **kwargs
149        )
150
151    async def delete_selfhosted_credentials(
152        self,
153        project_id: str,
154        distribution_credentials_id: str,
155        timeout: Optional[httpx.Timeout] = None,
156        **kwargs,
157    ):
158        """
159        Delete an on-premises distribution credential for a project.
160        """
161        self._logger.debug("SelfHostedClient.delete_selfhosted_credentials ENTER")
162        url = f"{self._config.url}/{self._endpoint}/{project_id}/selfhosted/distribution/credentials/{distribution_credentials_id}"
163        self._logger.info("url: %s", url)
164        self._logger.info("project_id: %s", project_id)
165        self._logger.info("distrbution_credentials_id: %s", distribution_credentials_id)
166        res = await self.delete(url, timeout=timeout, **kwargs)
167        self._logger.verbose("result: %s", res)
168        self._logger.notice("delete_selfhosted_credentials succeeded")
169        self._logger.debug("SelfHostedClient.delete_selfhosted_credentials LEAVE")
170        return res