deepgram.clients.selfhosted.client

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

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