deepgram.clients.analyze.v1.client

View Source

  1# Copyright 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 Dict, Union, Optional
  7
  8import httpx
  9
 10from ....utils import verboselogs
 11from ....options import DeepgramClientOptions
 12from ...common import AbstractSyncRestClient
 13from ...common import DeepgramError, DeepgramTypeError
 14
 15from .helpers import is_buffer_source, is_readstream_source, is_url_source
 16from .options import (
 17    AnalyzeOptions,
 18    UrlSource,
 19    FileSource,
 20)
 21from .response import AsyncAnalyzeResponse, AnalyzeResponse
 22
 23
 24class AnalyzeClient(AbstractSyncRestClient):
 25    """
 26    A client class for handling text data.
 27    Provides methods for transcribing text from URLs, files, etc.
 28    """
 29
 30    _logger: verboselogs.VerboseLogger
 31    _config: DeepgramClientOptions
 32
 33    def __init__(self, config: DeepgramClientOptions):
 34        self._logger = verboselogs.VerboseLogger(__name__)
 35        self._logger.addHandler(logging.StreamHandler())
 36        self._logger.setLevel(config.verbose)
 37        self._config = config
 38        super().__init__(config)
 39
 40    # pylint: disable=too-many-positional-arguments
 41
 42    def analyze_url(
 43        self,
 44        source: UrlSource,
 45        options: Optional[Union[AnalyzeOptions, Dict]] = None,
 46        addons: Optional[Dict] = None,
 47        headers: Optional[Dict] = None,
 48        timeout: Optional[httpx.Timeout] = None,
 49        endpoint: str = "v1/read",
 50        **kwargs,
 51    ) -> Union[AnalyzeResponse, AsyncAnalyzeResponse]:
 52        """
 53        Analyze text from a URL source.
 54
 55        Args:
 56            source (UrlSource): The URL source of the text to ingest.
 57            options (AnalyzeOptions): Additional options for the ingest (default is None).
 58            endpoint (str): The API endpoint for the ingest (default is "v1/read").
 59
 60        Returns:
 61            AnalyzeResponse: An object containing the result.
 62
 63        Raises:
 64            DeepgramTypeError: Raised for known API errors.
 65        """
 66        self._logger.debug("AnalyzeClient.analyze_url ENTER")
 67
 68        if (
 69            isinstance(options, dict)
 70            and "callback" in options
 71            and options["callback"] is not None
 72        ) or (isinstance(options, AnalyzeOptions) and options.callback is not None):
 73            self._logger.debug("AnalyzeClient.analyze_url LEAVE")
 74            return self.analyze_url_callback(
 75                source,
 76                callback=options["callback"],
 77                options=options,
 78                addons=addons,
 79                headers=headers,
 80                timeout=timeout,
 81                endpoint=endpoint,
 82                **kwargs,
 83            )
 84
 85        url = f"{self._config.url}/{endpoint}"
 86        if is_url_source(source):
 87            body = source
 88        else:
 89            self._logger.error("Unknown transcription source type")
 90            self._logger.debug("AnalyzeClient.analyze_url LEAVE")
 91            raise DeepgramTypeError("Unknown transcription source type")
 92
 93        if isinstance(options, AnalyzeOptions) and not options.check():
 94            self._logger.error("options.check failed")
 95            self._logger.debug("AnalyzeClient.analyze_url LEAVE")
 96            raise DeepgramError("Fatal transcription options error")
 97
 98        self._logger.info("url: %s", url)
 99        self._logger.info("source: %s", source)
100        if isinstance(options, AnalyzeOptions):
101            self._logger.info("AnalyzeOptions switching class -> dict")
102            options = options.to_dict()
103        self._logger.info("options: %s", options)
104        self._logger.info("addons: %s", addons)
105        self._logger.info("headers: %s", headers)
106        result = self.post(
107            url,
108            options=options,
109            addons=addons,
110            headers=headers,
111            json=body,
112            timeout=timeout,
113            **kwargs,
114        )
115        self._logger.info("json: %s", result)
116        res = AnalyzeResponse.from_json(result)
117        self._logger.verbose("result: %s", res)
118        self._logger.notice("analyze_url succeeded")
119        self._logger.debug("AnalyzeClient.analyze_url LEAVE")
120        return res
121
122    def analyze_url_callback(
123        self,
124        source: UrlSource,
125        callback: str,
126        options: Optional[Union[AnalyzeOptions, Dict]] = None,
127        addons: Optional[Dict] = None,
128        headers: Optional[Dict] = None,
129        timeout: Optional[httpx.Timeout] = None,
130        endpoint: str = "v1/read",
131        **kwargs,
132    ) -> AsyncAnalyzeResponse:
133        """
134        Transcribes audio from a URL source and sends the result to a callback URL.
135
136        Args:
137            source (UrlSource): The URL source of the audio to transcribe.
138            callback (str): The callback URL where the transcription results will be sent.
139            options (AnalyzeOptions): Additional options for the transcription (default is None).
140            endpoint (str): The API endpoint for the transcription (default is "v1/read").
141
142        Returns:
143            AsyncAnalyzeResponse: An object containing the request_id or an error message.
144
145        Raises:
146            DeepgramTypeError: Raised for known API errors.
147        """
148        self._logger.debug("AnalyzeClient.analyze_url_callback ENTER")
149
150        url = f"{self._config.url}/{endpoint}"
151        if options is None:
152            options = {}
153        if isinstance(options, AnalyzeOptions):
154            options.callback = callback
155        else:
156            options["callback"] = callback
157        if is_url_source(source):
158            body = source
159        else:
160            self._logger.error("Unknown transcription source type")
161            self._logger.debug("AnalyzeClient.analyze_url_callback LEAVE")
162            raise DeepgramTypeError("Unknown transcription source type")
163
164        if isinstance(options, AnalyzeOptions) and not options.check():
165            self._logger.error("options.check failed")
166            self._logger.debug("AnalyzeClient.analyze_url_callback LEAVE")
167            raise DeepgramError("Fatal transcription options error")
168
169        self._logger.info("url: %s", url)
170        self._logger.info("source: %s", source)
171        if isinstance(options, AnalyzeOptions):
172            self._logger.info("AnalyzeOptions switching class -> dict")
173            options = options.to_dict()
174        self._logger.info("options: %s", options)
175        self._logger.info("addons: %s", addons)
176        self._logger.info("headers: %s", headers)
177        result = self.post(
178            url,
179            options=options,
180            addons=addons,
181            headers=headers,
182            json=body,
183            timeout=timeout,
184            **kwargs,
185        )
186        self._logger.info("json: %s", result)
187        res = AsyncAnalyzeResponse.from_json(result)
188        self._logger.verbose("result: %s", res)
189        self._logger.notice("analyze_url_callback succeeded")
190        self._logger.debug("AnalyzeClient.analyze_url_callback LEAVE")
191        return res
192
193    def analyze_text(
194        self,
195        source: FileSource,
196        options: Optional[Union[AnalyzeOptions, Dict]] = None,
197        addons: Optional[Dict] = None,
198        headers: Optional[Dict] = None,
199        timeout: Optional[httpx.Timeout] = None,
200        endpoint: str = "v1/read",
201        **kwargs,
202    ) -> Union[AnalyzeResponse, AsyncAnalyzeResponse]:
203        """
204        Analyze text from a local file source.
205
206        Args:
207            source (TextSource): The local file source of the text to ingest.
208            options (AnalyzeOptions): Additional options for the ingest (default is None).
209            endpoint (str): The API endpoint for the transcription (default is "v1/read").
210
211        Returns:
212            AnalyzeResponse: An object containing the transcription result or an error message.
213
214        Raises:
215            DeepgramTypeError: Raised for known API errors.
216        """
217        self._logger.debug("AnalyzeClient.analyze_text ENTER")
218
219        if (
220            isinstance(options, dict)
221            and "callback" in options
222            and options["callback"] is not None
223        ) or (isinstance(options, AnalyzeOptions) and options.callback is not None):
224            self._logger.debug("AnalyzeClient.analyze_text LEAVE")
225            return self.analyze_text_callback(
226                source,
227                callback=options["callback"],
228                options=options,
229                addons=addons,
230                headers=headers,
231                timeout=timeout,
232                endpoint=endpoint,
233                **kwargs,
234            )
235
236        url = f"{self._config.url}/{endpoint}"
237        if is_buffer_source(source):
238            body = source["buffer"]  # type: ignore
239        elif is_readstream_source(source):
240            body = source["stream"]  # type: ignore
241        else:
242            self._logger.error("Unknown transcription source type")
243            self._logger.debug("AnalyzeClient.analyze_text LEAVE")
244            raise DeepgramTypeError("Unknown transcription source type")
245
246        if isinstance(options, AnalyzeOptions) and not options.check():
247            self._logger.error("options.check failed")
248            self._logger.debug("AnalyzeClient.analyze_text LEAVE")
249            raise DeepgramError("Fatal transcription options error")
250
251        self._logger.info("url: %s", url)
252        if isinstance(options, AnalyzeOptions):
253            self._logger.info("AnalyzeOptions switching class -> dict")
254            options = options.to_dict()
255        self._logger.info("options: %s", options)
256        self._logger.info("addons: %s", addons)
257        self._logger.info("headers: %s", headers)
258        result = self.post(
259            url,
260            options=options,
261            addons=addons,
262            headers=headers,
263            content=body,
264            timeout=timeout,
265            **kwargs,
266        )
267        self._logger.info("json: %s", result)
268        res = AnalyzeResponse.from_json(result)
269        self._logger.verbose("result: %s", res)
270        self._logger.notice("analyze_text succeeded")
271        self._logger.debug("AnalyzeClient.analyze_text LEAVE")
272        return res
273
274    def analyze_text_callback(
275        self,
276        source: FileSource,
277        callback: str,
278        options: Optional[Union[AnalyzeOptions, Dict]] = None,
279        addons: Optional[Dict] = None,
280        headers: Optional[Dict] = None,
281        timeout: Optional[httpx.Timeout] = None,
282        endpoint: str = "v1/read",
283        **kwargs,
284    ) -> AsyncAnalyzeResponse:
285        """
286        Transcribes audio from a local file source and sends the result to a callback URL.
287
288        Args:
289            source (TextSource): The local file source of the audio to transcribe.
290            callback (str): The callback URL where the transcription results will be sent.
291            options (AnalyzeOptions): Additional options for the transcription (default is None).
292            endpoint (str): The API endpoint for the transcription (default is "v1/read").
293
294        Returns:
295            AsyncAnalyzeResponse: An object containing the request_id or an error message.
296
297        Raises:
298            DeepgramTypeError: Raised for known API errors.
299        """
300        self._logger.debug("AnalyzeClient.analyze_file_callback ENTER")
301
302        url = f"{self._config.url}/{endpoint}"
303        if options is None:
304            options = {}
305        if isinstance(options, AnalyzeOptions):
306            options.callback = callback
307        else:
308            options["callback"] = callback
309        if is_buffer_source(source):
310            body = source["buffer"]  # type: ignore
311        elif is_readstream_source(source):
312            body = source["stream"]  # type: ignore
313        else:
314            self._logger.error("Unknown transcription source type")
315            self._logger.debug("AnalyzeClient.analyze_file_callback LEAVE")
316            raise DeepgramTypeError("Unknown transcription source type")
317
318        if isinstance(options, AnalyzeOptions) and not options.check():
319            self._logger.error("options.check failed")
320            self._logger.debug("AnalyzeClient.analyze_file_callback LEAVE")
321            raise DeepgramError("Fatal transcription options error")
322
323        self._logger.info("url: %s", url)
324        if isinstance(options, AnalyzeOptions):
325            self._logger.info("AnalyzeOptions switching class -> dict")
326            options = options.to_dict()
327        self._logger.info("options: %s", options)
328        self._logger.info("addons: %s", addons)
329        self._logger.info("headers: %s", headers)
330        result = self.post(
331            url,
332            options=options,
333            addons=addons,
334            headers=headers,
335            json=body,
336            timeout=timeout,
337            **kwargs,
338        )
339        self._logger.info("json: %s", result)
340        res = AsyncAnalyzeResponse.from_json(result)
341        self._logger.verbose("result: %s", res)
342        self._logger.notice("analyze_file_callback succeeded")
343        self._logger.debug("AnalyzeClient.analyze_file_callback LEAVE")
344        return res
345
346    # pylint: enable=too-many-positional-arguments

class AnalyzeClient(deepgram.clients.common.v1.abstract_sync_rest.AbstractSyncRestClient): View Source

 25class AnalyzeClient(AbstractSyncRestClient):
 26    """
 27    A client class for handling text data.
 28    Provides methods for transcribing text from URLs, files, etc.
 29    """
 30
 31    _logger: verboselogs.VerboseLogger
 32    _config: DeepgramClientOptions
 33
 34    def __init__(self, config: DeepgramClientOptions):
 35        self._logger = verboselogs.VerboseLogger(__name__)
 36        self._logger.addHandler(logging.StreamHandler())
 37        self._logger.setLevel(config.verbose)
 38        self._config = config
 39        super().__init__(config)
 40
 41    # pylint: disable=too-many-positional-arguments
 42
 43    def analyze_url(
 44        self,
 45        source: UrlSource,
 46        options: Optional[Union[AnalyzeOptions, Dict]] = None,
 47        addons: Optional[Dict] = None,
 48        headers: Optional[Dict] = None,
 49        timeout: Optional[httpx.Timeout] = None,
 50        endpoint: str = "v1/read",
 51        **kwargs,
 52    ) -> Union[AnalyzeResponse, AsyncAnalyzeResponse]:
 53        """
 54        Analyze text from a URL source.
 55
 56        Args:
 57            source (UrlSource): The URL source of the text to ingest.
 58            options (AnalyzeOptions): Additional options for the ingest (default is None).
 59            endpoint (str): The API endpoint for the ingest (default is "v1/read").
 60
 61        Returns:
 62            AnalyzeResponse: An object containing the result.
 63
 64        Raises:
 65            DeepgramTypeError: Raised for known API errors.
 66        """
 67        self._logger.debug("AnalyzeClient.analyze_url ENTER")
 68
 69        if (
 70            isinstance(options, dict)
 71            and "callback" in options
 72            and options["callback"] is not None
 73        ) or (isinstance(options, AnalyzeOptions) and options.callback is not None):
 74            self._logger.debug("AnalyzeClient.analyze_url LEAVE")
 75            return self.analyze_url_callback(
 76                source,
 77                callback=options["callback"],
 78                options=options,
 79                addons=addons,
 80                headers=headers,
 81                timeout=timeout,
 82                endpoint=endpoint,
 83                **kwargs,
 84            )
 85
 86        url = f"{self._config.url}/{endpoint}"
 87        if is_url_source(source):
 88            body = source
 89        else:
 90            self._logger.error("Unknown transcription source type")
 91            self._logger.debug("AnalyzeClient.analyze_url LEAVE")
 92            raise DeepgramTypeError("Unknown transcription source type")
 93
 94        if isinstance(options, AnalyzeOptions) and not options.check():
 95            self._logger.error("options.check failed")
 96            self._logger.debug("AnalyzeClient.analyze_url LEAVE")
 97            raise DeepgramError("Fatal transcription options error")
 98
 99        self._logger.info("url: %s", url)
100        self._logger.info("source: %s", source)
101        if isinstance(options, AnalyzeOptions):
102            self._logger.info("AnalyzeOptions switching class -> dict")
103            options = options.to_dict()
104        self._logger.info("options: %s", options)
105        self._logger.info("addons: %s", addons)
106        self._logger.info("headers: %s", headers)
107        result = self.post(
108            url,
109            options=options,
110            addons=addons,
111            headers=headers,
112            json=body,
113            timeout=timeout,
114            **kwargs,
115        )
116        self._logger.info("json: %s", result)
117        res = AnalyzeResponse.from_json(result)
118        self._logger.verbose("result: %s", res)
119        self._logger.notice("analyze_url succeeded")
120        self._logger.debug("AnalyzeClient.analyze_url LEAVE")
121        return res
122
123    def analyze_url_callback(
124        self,
125        source: UrlSource,
126        callback: str,
127        options: Optional[Union[AnalyzeOptions, Dict]] = None,
128        addons: Optional[Dict] = None,
129        headers: Optional[Dict] = None,
130        timeout: Optional[httpx.Timeout] = None,
131        endpoint: str = "v1/read",
132        **kwargs,
133    ) -> AsyncAnalyzeResponse:
134        """
135        Transcribes audio from a URL source and sends the result to a callback URL.
136
137        Args:
138            source (UrlSource): The URL source of the audio to transcribe.
139            callback (str): The callback URL where the transcription results will be sent.
140            options (AnalyzeOptions): Additional options for the transcription (default is None).
141            endpoint (str): The API endpoint for the transcription (default is "v1/read").
142
143        Returns:
144            AsyncAnalyzeResponse: An object containing the request_id or an error message.
145
146        Raises:
147            DeepgramTypeError: Raised for known API errors.
148        """
149        self._logger.debug("AnalyzeClient.analyze_url_callback ENTER")
150
151        url = f"{self._config.url}/{endpoint}"
152        if options is None:
153            options = {}
154        if isinstance(options, AnalyzeOptions):
155            options.callback = callback
156        else:
157            options["callback"] = callback
158        if is_url_source(source):
159            body = source
160        else:
161            self._logger.error("Unknown transcription source type")
162            self._logger.debug("AnalyzeClient.analyze_url_callback LEAVE")
163            raise DeepgramTypeError("Unknown transcription source type")
164
165        if isinstance(options, AnalyzeOptions) and not options.check():
166            self._logger.error("options.check failed")
167            self._logger.debug("AnalyzeClient.analyze_url_callback LEAVE")
168            raise DeepgramError("Fatal transcription options error")
169
170        self._logger.info("url: %s", url)
171        self._logger.info("source: %s", source)
172        if isinstance(options, AnalyzeOptions):
173            self._logger.info("AnalyzeOptions switching class -> dict")
174            options = options.to_dict()
175        self._logger.info("options: %s", options)
176        self._logger.info("addons: %s", addons)
177        self._logger.info("headers: %s", headers)
178        result = self.post(
179            url,
180            options=options,
181            addons=addons,
182            headers=headers,
183            json=body,
184            timeout=timeout,
185            **kwargs,
186        )
187        self._logger.info("json: %s", result)
188        res = AsyncAnalyzeResponse.from_json(result)
189        self._logger.verbose("result: %s", res)
190        self._logger.notice("analyze_url_callback succeeded")
191        self._logger.debug("AnalyzeClient.analyze_url_callback LEAVE")
192        return res
193
194    def analyze_text(
195        self,
196        source: FileSource,
197        options: Optional[Union[AnalyzeOptions, Dict]] = None,
198        addons: Optional[Dict] = None,
199        headers: Optional[Dict] = None,
200        timeout: Optional[httpx.Timeout] = None,
201        endpoint: str = "v1/read",
202        **kwargs,
203    ) -> Union[AnalyzeResponse, AsyncAnalyzeResponse]:
204        """
205        Analyze text from a local file source.
206
207        Args:
208            source (TextSource): The local file source of the text to ingest.
209            options (AnalyzeOptions): Additional options for the ingest (default is None).
210            endpoint (str): The API endpoint for the transcription (default is "v1/read").
211
212        Returns:
213            AnalyzeResponse: An object containing the transcription result or an error message.
214
215        Raises:
216            DeepgramTypeError: Raised for known API errors.
217        """
218        self._logger.debug("AnalyzeClient.analyze_text ENTER")
219
220        if (
221            isinstance(options, dict)
222            and "callback" in options
223            and options["callback"] is not None
224        ) or (isinstance(options, AnalyzeOptions) and options.callback is not None):
225            self._logger.debug("AnalyzeClient.analyze_text LEAVE")
226            return self.analyze_text_callback(
227                source,
228                callback=options["callback"],
229                options=options,
230                addons=addons,
231                headers=headers,
232                timeout=timeout,
233                endpoint=endpoint,
234                **kwargs,
235            )
236
237        url = f"{self._config.url}/{endpoint}"
238        if is_buffer_source(source):
239            body = source["buffer"]  # type: ignore
240        elif is_readstream_source(source):
241            body = source["stream"]  # type: ignore
242        else:
243            self._logger.error("Unknown transcription source type")
244            self._logger.debug("AnalyzeClient.analyze_text LEAVE")
245            raise DeepgramTypeError("Unknown transcription source type")
246
247        if isinstance(options, AnalyzeOptions) and not options.check():
248            self._logger.error("options.check failed")
249            self._logger.debug("AnalyzeClient.analyze_text LEAVE")
250            raise DeepgramError("Fatal transcription options error")
251
252        self._logger.info("url: %s", url)
253        if isinstance(options, AnalyzeOptions):
254            self._logger.info("AnalyzeOptions switching class -> dict")
255            options = options.to_dict()
256        self._logger.info("options: %s", options)
257        self._logger.info("addons: %s", addons)
258        self._logger.info("headers: %s", headers)
259        result = self.post(
260            url,
261            options=options,
262            addons=addons,
263            headers=headers,
264            content=body,
265            timeout=timeout,
266            **kwargs,
267        )
268        self._logger.info("json: %s", result)
269        res = AnalyzeResponse.from_json(result)
270        self._logger.verbose("result: %s", res)
271        self._logger.notice("analyze_text succeeded")
272        self._logger.debug("AnalyzeClient.analyze_text LEAVE")
273        return res
274
275    def analyze_text_callback(
276        self,
277        source: FileSource,
278        callback: str,
279        options: Optional[Union[AnalyzeOptions, Dict]] = None,
280        addons: Optional[Dict] = None,
281        headers: Optional[Dict] = None,
282        timeout: Optional[httpx.Timeout] = None,
283        endpoint: str = "v1/read",
284        **kwargs,
285    ) -> AsyncAnalyzeResponse:
286        """
287        Transcribes audio from a local file source and sends the result to a callback URL.
288
289        Args:
290            source (TextSource): The local file source of the audio to transcribe.
291            callback (str): The callback URL where the transcription results will be sent.
292            options (AnalyzeOptions): Additional options for the transcription (default is None).
293            endpoint (str): The API endpoint for the transcription (default is "v1/read").
294
295        Returns:
296            AsyncAnalyzeResponse: An object containing the request_id or an error message.
297
298        Raises:
299            DeepgramTypeError: Raised for known API errors.
300        """
301        self._logger.debug("AnalyzeClient.analyze_file_callback ENTER")
302
303        url = f"{self._config.url}/{endpoint}"
304        if options is None:
305            options = {}
306        if isinstance(options, AnalyzeOptions):
307            options.callback = callback
308        else:
309            options["callback"] = callback
310        if is_buffer_source(source):
311            body = source["buffer"]  # type: ignore
312        elif is_readstream_source(source):
313            body = source["stream"]  # type: ignore
314        else:
315            self._logger.error("Unknown transcription source type")
316            self._logger.debug("AnalyzeClient.analyze_file_callback LEAVE")
317            raise DeepgramTypeError("Unknown transcription source type")
318
319        if isinstance(options, AnalyzeOptions) and not options.check():
320            self._logger.error("options.check failed")
321            self._logger.debug("AnalyzeClient.analyze_file_callback LEAVE")
322            raise DeepgramError("Fatal transcription options error")
323
324        self._logger.info("url: %s", url)
325        if isinstance(options, AnalyzeOptions):
326            self._logger.info("AnalyzeOptions switching class -> dict")
327            options = options.to_dict()
328        self._logger.info("options: %s", options)
329        self._logger.info("addons: %s", addons)
330        self._logger.info("headers: %s", headers)
331        result = self.post(
332            url,
333            options=options,
334            addons=addons,
335            headers=headers,
336            json=body,
337            timeout=timeout,
338            **kwargs,
339        )
340        self._logger.info("json: %s", result)
341        res = AsyncAnalyzeResponse.from_json(result)
342        self._logger.verbose("result: %s", res)
343        self._logger.notice("analyze_file_callback succeeded")
344        self._logger.debug("AnalyzeClient.analyze_file_callback LEAVE")
345        return res
346
347    # pylint: enable=too-many-positional-arguments

A client class for handling text data. Provides methods for transcribing text from URLs, files, etc.

AnalyzeClient(config: deepgram.options.DeepgramClientOptions) View Source

34    def __init__(self, config: DeepgramClientOptions):
35        self._logger = verboselogs.VerboseLogger(__name__)
36        self._logger.addHandler(logging.StreamHandler())
37        self._logger.setLevel(config.verbose)
38        self._config = config
39        super().__init__(config)

def analyze_url( self, source: deepgram.clients.common.UrlSource, options: Union[deepgram.clients.analyze.v1.options.AnalyzeOptions, Dict, NoneType] = None, addons: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: Optional[httpx.Timeout] = None, endpoint: str = 'v1/read', **kwargs) -> Union[deepgram.clients.analyze.v1.response.AnalyzeResponse, deepgram.clients.analyze.v1.response.AsyncAnalyzeResponse]: View Source

 43    def analyze_url(
 44        self,
 45        source: UrlSource,
 46        options: Optional[Union[AnalyzeOptions, Dict]] = None,
 47        addons: Optional[Dict] = None,
 48        headers: Optional[Dict] = None,
 49        timeout: Optional[httpx.Timeout] = None,
 50        endpoint: str = "v1/read",
 51        **kwargs,
 52    ) -> Union[AnalyzeResponse, AsyncAnalyzeResponse]:
 53        """
 54        Analyze text from a URL source.
 55
 56        Args:
 57            source (UrlSource): The URL source of the text to ingest.
 58            options (AnalyzeOptions): Additional options for the ingest (default is None).
 59            endpoint (str): The API endpoint for the ingest (default is "v1/read").
 60
 61        Returns:
 62            AnalyzeResponse: An object containing the result.
 63
 64        Raises:
 65            DeepgramTypeError: Raised for known API errors.
 66        """
 67        self._logger.debug("AnalyzeClient.analyze_url ENTER")
 68
 69        if (
 70            isinstance(options, dict)
 71            and "callback" in options
 72            and options["callback"] is not None
 73        ) or (isinstance(options, AnalyzeOptions) and options.callback is not None):
 74            self._logger.debug("AnalyzeClient.analyze_url LEAVE")
 75            return self.analyze_url_callback(
 76                source,
 77                callback=options["callback"],
 78                options=options,
 79                addons=addons,
 80                headers=headers,
 81                timeout=timeout,
 82                endpoint=endpoint,
 83                **kwargs,
 84            )
 85
 86        url = f"{self._config.url}/{endpoint}"
 87        if is_url_source(source):
 88            body = source
 89        else:
 90            self._logger.error("Unknown transcription source type")
 91            self._logger.debug("AnalyzeClient.analyze_url LEAVE")
 92            raise DeepgramTypeError("Unknown transcription source type")
 93
 94        if isinstance(options, AnalyzeOptions) and not options.check():
 95            self._logger.error("options.check failed")
 96            self._logger.debug("AnalyzeClient.analyze_url LEAVE")
 97            raise DeepgramError("Fatal transcription options error")
 98
 99        self._logger.info("url: %s", url)
100        self._logger.info("source: %s", source)
101        if isinstance(options, AnalyzeOptions):
102            self._logger.info("AnalyzeOptions switching class -> dict")
103            options = options.to_dict()
104        self._logger.info("options: %s", options)
105        self._logger.info("addons: %s", addons)
106        self._logger.info("headers: %s", headers)
107        result = self.post(
108            url,
109            options=options,
110            addons=addons,
111            headers=headers,
112            json=body,
113            timeout=timeout,
114            **kwargs,
115        )
116        self._logger.info("json: %s", result)
117        res = AnalyzeResponse.from_json(result)
118        self._logger.verbose("result: %s", res)
119        self._logger.notice("analyze_url succeeded")
120        self._logger.debug("AnalyzeClient.analyze_url LEAVE")
121        return res

Analyze text from a URL source.

Args: source (UrlSource): The URL source of the text to ingest. options (AnalyzeOptions): Additional options for the ingest (default is None). endpoint (str): The API endpoint for the ingest (default is "v1/read").

Returns: AnalyzeResponse: An object containing the result.

Raises: DeepgramTypeError: Raised for known API errors.

def analyze_url_callback( self, source: deepgram.clients.common.UrlSource, callback: str, options: Union[deepgram.clients.analyze.v1.options.AnalyzeOptions, Dict, NoneType] = None, addons: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: Optional[httpx.Timeout] = None, endpoint: str = 'v1/read', **kwargs) -> deepgram.clients.analyze.v1.response.AsyncAnalyzeResponse: View Source

123    def analyze_url_callback(
124        self,
125        source: UrlSource,
126        callback: str,
127        options: Optional[Union[AnalyzeOptions, Dict]] = None,
128        addons: Optional[Dict] = None,
129        headers: Optional[Dict] = None,
130        timeout: Optional[httpx.Timeout] = None,
131        endpoint: str = "v1/read",
132        **kwargs,
133    ) -> AsyncAnalyzeResponse:
134        """
135        Transcribes audio from a URL source and sends the result to a callback URL.
136
137        Args:
138            source (UrlSource): The URL source of the audio to transcribe.
139            callback (str): The callback URL where the transcription results will be sent.
140            options (AnalyzeOptions): Additional options for the transcription (default is None).
141            endpoint (str): The API endpoint for the transcription (default is "v1/read").
142
143        Returns:
144            AsyncAnalyzeResponse: An object containing the request_id or an error message.
145
146        Raises:
147            DeepgramTypeError: Raised for known API errors.
148        """
149        self._logger.debug("AnalyzeClient.analyze_url_callback ENTER")
150
151        url = f"{self._config.url}/{endpoint}"
152        if options is None:
153            options = {}
154        if isinstance(options, AnalyzeOptions):
155            options.callback = callback
156        else:
157            options["callback"] = callback
158        if is_url_source(source):
159            body = source
160        else:
161            self._logger.error("Unknown transcription source type")
162            self._logger.debug("AnalyzeClient.analyze_url_callback LEAVE")
163            raise DeepgramTypeError("Unknown transcription source type")
164
165        if isinstance(options, AnalyzeOptions) and not options.check():
166            self._logger.error("options.check failed")
167            self._logger.debug("AnalyzeClient.analyze_url_callback LEAVE")
168            raise DeepgramError("Fatal transcription options error")
169
170        self._logger.info("url: %s", url)
171        self._logger.info("source: %s", source)
172        if isinstance(options, AnalyzeOptions):
173            self._logger.info("AnalyzeOptions switching class -> dict")
174            options = options.to_dict()
175        self._logger.info("options: %s", options)
176        self._logger.info("addons: %s", addons)
177        self._logger.info("headers: %s", headers)
178        result = self.post(
179            url,
180            options=options,
181            addons=addons,
182            headers=headers,
183            json=body,
184            timeout=timeout,
185            **kwargs,
186        )
187        self._logger.info("json: %s", result)
188        res = AsyncAnalyzeResponse.from_json(result)
189        self._logger.verbose("result: %s", res)
190        self._logger.notice("analyze_url_callback succeeded")
191        self._logger.debug("AnalyzeClient.analyze_url_callback LEAVE")
192        return res

Transcribes audio from a URL source and sends the result to a callback URL.

Args: source (UrlSource): The URL source of the audio to transcribe. callback (str): The callback URL where the transcription results will be sent. options (AnalyzeOptions): Additional options for the transcription (default is None). endpoint (str): The API endpoint for the transcription (default is "v1/read").

Returns: AsyncAnalyzeResponse: An object containing the request_id or an error message.

Raises: DeepgramTypeError: Raised for known API errors.

def analyze_text( self, source: Union[deepgram.clients.common.TextSource, deepgram.clients.common.BufferSource, deepgram.clients.common.StreamSource], options: Union[deepgram.clients.analyze.v1.options.AnalyzeOptions, Dict, NoneType] = None, addons: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: Optional[httpx.Timeout] = None, endpoint: str = 'v1/read', **kwargs) -> Union[deepgram.clients.analyze.v1.response.AnalyzeResponse, deepgram.clients.analyze.v1.response.AsyncAnalyzeResponse]: View Source

194    def analyze_text(
195        self,
196        source: FileSource,
197        options: Optional[Union[AnalyzeOptions, Dict]] = None,
198        addons: Optional[Dict] = None,
199        headers: Optional[Dict] = None,
200        timeout: Optional[httpx.Timeout] = None,
201        endpoint: str = "v1/read",
202        **kwargs,
203    ) -> Union[AnalyzeResponse, AsyncAnalyzeResponse]:
204        """
205        Analyze text from a local file source.
206
207        Args:
208            source (TextSource): The local file source of the text to ingest.
209            options (AnalyzeOptions): Additional options for the ingest (default is None).
210            endpoint (str): The API endpoint for the transcription (default is "v1/read").
211
212        Returns:
213            AnalyzeResponse: An object containing the transcription result or an error message.
214
215        Raises:
216            DeepgramTypeError: Raised for known API errors.
217        """
218        self._logger.debug("AnalyzeClient.analyze_text ENTER")
219
220        if (
221            isinstance(options, dict)
222            and "callback" in options
223            and options["callback"] is not None
224        ) or (isinstance(options, AnalyzeOptions) and options.callback is not None):
225            self._logger.debug("AnalyzeClient.analyze_text LEAVE")
226            return self.analyze_text_callback(
227                source,
228                callback=options["callback"],
229                options=options,
230                addons=addons,
231                headers=headers,
232                timeout=timeout,
233                endpoint=endpoint,
234                **kwargs,
235            )
236
237        url = f"{self._config.url}/{endpoint}"
238        if is_buffer_source(source):
239            body = source["buffer"]  # type: ignore
240        elif is_readstream_source(source):
241            body = source["stream"]  # type: ignore
242        else:
243            self._logger.error("Unknown transcription source type")
244            self._logger.debug("AnalyzeClient.analyze_text LEAVE")
245            raise DeepgramTypeError("Unknown transcription source type")
246
247        if isinstance(options, AnalyzeOptions) and not options.check():
248            self._logger.error("options.check failed")
249            self._logger.debug("AnalyzeClient.analyze_text LEAVE")
250            raise DeepgramError("Fatal transcription options error")
251
252        self._logger.info("url: %s", url)
253        if isinstance(options, AnalyzeOptions):
254            self._logger.info("AnalyzeOptions switching class -> dict")
255            options = options.to_dict()
256        self._logger.info("options: %s", options)
257        self._logger.info("addons: %s", addons)
258        self._logger.info("headers: %s", headers)
259        result = self.post(
260            url,
261            options=options,
262            addons=addons,
263            headers=headers,
264            content=body,
265            timeout=timeout,
266            **kwargs,
267        )
268        self._logger.info("json: %s", result)
269        res = AnalyzeResponse.from_json(result)
270        self._logger.verbose("result: %s", res)
271        self._logger.notice("analyze_text succeeded")
272        self._logger.debug("AnalyzeClient.analyze_text LEAVE")
273        return res

Analyze text from a local file source.

Args: source (TextSource): The local file source of the text to ingest. options (AnalyzeOptions): Additional options for the ingest (default is None). endpoint (str): The API endpoint for the transcription (default is "v1/read").

Returns: AnalyzeResponse: An object containing the transcription result or an error message.

Raises: DeepgramTypeError: Raised for known API errors.

def analyze_text_callback( self, source: Union[deepgram.clients.common.TextSource, deepgram.clients.common.BufferSource, deepgram.clients.common.StreamSource], callback: str, options: Union[deepgram.clients.analyze.v1.options.AnalyzeOptions, Dict, NoneType] = None, addons: Optional[Dict] = None, headers: Optional[Dict] = None, timeout: Optional[httpx.Timeout] = None, endpoint: str = 'v1/read', **kwargs) -> deepgram.clients.analyze.v1.response.AsyncAnalyzeResponse: View Source

275    def analyze_text_callback(
276        self,
277        source: FileSource,
278        callback: str,
279        options: Optional[Union[AnalyzeOptions, Dict]] = None,
280        addons: Optional[Dict] = None,
281        headers: Optional[Dict] = None,
282        timeout: Optional[httpx.Timeout] = None,
283        endpoint: str = "v1/read",
284        **kwargs,
285    ) -> AsyncAnalyzeResponse:
286        """
287        Transcribes audio from a local file source and sends the result to a callback URL.
288
289        Args:
290            source (TextSource): The local file source of the audio to transcribe.
291            callback (str): The callback URL where the transcription results will be sent.
292            options (AnalyzeOptions): Additional options for the transcription (default is None).
293            endpoint (str): The API endpoint for the transcription (default is "v1/read").
294
295        Returns:
296            AsyncAnalyzeResponse: An object containing the request_id or an error message.
297
298        Raises:
299            DeepgramTypeError: Raised for known API errors.
300        """
301        self._logger.debug("AnalyzeClient.analyze_file_callback ENTER")
302
303        url = f"{self._config.url}/{endpoint}"
304        if options is None:
305            options = {}
306        if isinstance(options, AnalyzeOptions):
307            options.callback = callback
308        else:
309            options["callback"] = callback
310        if is_buffer_source(source):
311            body = source["buffer"]  # type: ignore
312        elif is_readstream_source(source):
313            body = source["stream"]  # type: ignore
314        else:
315            self._logger.error("Unknown transcription source type")
316            self._logger.debug("AnalyzeClient.analyze_file_callback LEAVE")
317            raise DeepgramTypeError("Unknown transcription source type")
318
319        if isinstance(options, AnalyzeOptions) and not options.check():
320            self._logger.error("options.check failed")
321            self._logger.debug("AnalyzeClient.analyze_file_callback LEAVE")
322            raise DeepgramError("Fatal transcription options error")
323
324        self._logger.info("url: %s", url)
325        if isinstance(options, AnalyzeOptions):
326            self._logger.info("AnalyzeOptions switching class -> dict")
327            options = options.to_dict()
328        self._logger.info("options: %s", options)
329        self._logger.info("addons: %s", addons)
330        self._logger.info("headers: %s", headers)
331        result = self.post(
332            url,
333            options=options,
334            addons=addons,
335            headers=headers,
336            json=body,
337            timeout=timeout,
338            **kwargs,
339        )
340        self._logger.info("json: %s", result)
341        res = AsyncAnalyzeResponse.from_json(result)
342        self._logger.verbose("result: %s", res)
343        self._logger.notice("analyze_file_callback succeeded")
344        self._logger.debug("AnalyzeClient.analyze_file_callback LEAVE")
345        return res

Transcribes audio from a local file source and sends the result to a callback URL.

Args: source (TextSource): The local file source of the audio to transcribe. callback (str): The callback URL where the transcription results will be sent. options (AnalyzeOptions): Additional options for the transcription (default is None). endpoint (str): The API endpoint for the transcription (default is "v1/read").

Returns: AsyncAnalyzeResponse: An object containing the request_id or an error message.

Raises: DeepgramTypeError: Raised for known API errors.

Inherited Members

deepgram.clients.common.v1.abstract_sync_rest.AbstractSyncRestClient: get; post_raw; post_memory; post; put; patch; delete