deepgram.clients.selfhosted.v1.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 AbstractSyncRestClient
 13
 14
 15class SelfHostedClient(AbstractSyncRestClient):
 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    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    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 = 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    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    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 = 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    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_selfhosted_credentials(
115            project_id, options, timeout=timeout, **kwargs
116        )
117
118    def create_selfhosted_credentials(
119        self,
120        project_id: str,
121        options,
122        timeout: Optional[httpx.Timeout] = None,
123        **kwargs,
124    ):
125        """
126        Create a new on-premises distribution credential for a project.
127        """
128        self._logger.debug("SelfHostedClient.create_selfhosted_credentials ENTER")
129        url = f"{self._config.url}/{self._endpoint}/{project_id}/selfhosted/distribution/credentials/"
130        self._logger.info("url: %s", url)
131        self._logger.info("project_id: %s", project_id)
132        self._logger.info("options: %s", options)
133        res = self.post(url, json=options, timeout=timeout, **kwargs)
134        self._logger.verbose("result: %s", res)
135        self._logger.notice("create_selfhosted_credentials succeeded")
136        self._logger.debug("SelfHostedClient.create_selfhosted_credentials LEAVE")
137        return res
138
139    def delete_onprem_credentials(
140        self,
141        project_id: str,
142        distribution_credentials_id: str,
143        timeout: Optional[httpx.Timeout] = None,
144        **kwargs,
145    ):
146        """
147        Delete an on-premises distribution credential for a project.
148        """
149        return self.delete_selfhosted_credentials(
150            project_id, distribution_credentials_id, timeout=timeout, **kwargs
151        )
152
153    def delete_selfhosted_credentials(
154        self,
155        project_id: str,
156        distribution_credentials_id: str,
157        timeout: Optional[httpx.Timeout] = None,
158        **kwargs,
159    ):
160        """
161        Delete an on-premises distribution credential for a project.
162        """
163        self._logger.debug("SelfHostedClient.delete_selfhosted_credentials ENTER")
164        url = f"{self._config.url}/{self._endpoint}/{project_id}/selfhosted/distribution/credentials/{distribution_credentials_id}"
165        self._logger.info("url: %s", url)
166        self._logger.info("project_id: %s", project_id)
167        self._logger.info("distrbution_credentials_id: %s", distribution_credentials_id)
168        res = self.delete(url, timeout=timeout, **kwargs)
169        self._logger.verbose("result: %s", res)
170        self._logger.notice("delete_selfhosted_credentials succeeded")
171        self._logger.debug("SelfHostedClient.delete_selfhosted_credentials LEAVE")
172        return res