deepgram.options
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 sys 6import re 7import os 8from typing import Dict, Optional 9import logging 10import numbers 11 12from deepgram import __version__ 13from .utils import verboselogs 14from .errors import DeepgramApiKeyError 15 16 17class DeepgramClientOptions: # pylint: disable=too-many-instance-attributes 18 """ 19 Represents options for configuring a Deepgram client. 20 21 This class allows you to customize various options for interacting with the Deepgram API. 22 23 Attributes: 24 api_key: (Optional) A Deepgram API key used for authentication. Default uses the `DEEPGRAM_API_KEY` environment variable. 25 url: (Optional) The URL used to interact with production, On-prem, and other Deepgram environments. Defaults to `api.deepgram.com`. 26 verbose: (Optional) The logging level for the client. Defaults to `verboselogs.WARNING`. 27 headers: (Optional) Headers for initializing the client. 28 options: (Optional) Additional options for initializing the client. 29 """ 30 31 _logger: verboselogs.VerboseLogger 32 _inspect_listen: bool = False 33 _inspect_speak: bool = False 34 35 def __init__( 36 self, 37 api_key: str = "", 38 url: str = "", 39 verbose: int = verboselogs.WARNING, 40 headers: Optional[Dict] = None, 41 options: Optional[Dict] = None, 42 ): # pylint: disable=too-many-positional-arguments 43 self._logger = verboselogs.VerboseLogger(__name__) 44 self._logger.addHandler(logging.StreamHandler()) 45 46 if api_key is None: 47 api_key = "" 48 49 self.verbose = verbose 50 self.api_key = api_key 51 52 if headers is None: 53 headers = {} 54 self._update_headers(headers=headers) 55 56 if len(url) == 0: 57 url = "api.deepgram.com" 58 self.url = self._get_url(url) 59 60 if options is None: 61 options = {} 62 self.options = options 63 64 if self.is_auto_flush_reply_enabled(): 65 self._inspect_listen = True 66 if self.is_auto_flush_speak_enabled(): 67 self._inspect_speak = True 68 69 def set_apikey(self, api_key: str): 70 """ 71 set_apikey: Sets the API key for the client. 72 73 Args: 74 api_key: The Deepgram API key used for authentication. 75 """ 76 self.api_key = api_key 77 self._update_headers() 78 79 def _get_url(self, url) -> str: 80 if not re.match(r"^https?://", url, re.IGNORECASE): 81 url = "https://" + url 82 return url.strip("/") 83 84 def _update_headers(self, headers: Optional[Dict] = None): 85 self.headers = {} 86 self.headers["Accept"] = "application/json" 87 if self.api_key: 88 self.headers["Authorization"] = f"Token {self.api_key}" 89 elif "Authorization" in self.headers: 90 del self.headers["Authorization"] 91 self.headers[ 92 "User-Agent" 93 ] = f"@deepgram/sdk/{__version__} python/{sys.version_info[1]}.{sys.version_info[2]}" 94 # Overwrite / add any headers that were passed in 95 if headers: 96 self.headers.update(headers) 97 98 def is_keep_alive_enabled(self) -> bool: 99 """ 100 is_keep_alive_enabled: Returns True if the client is configured to keep the connection alive. 101 """ 102 return self.options.get("keepalive", False) or self.options.get( 103 "keep_alive", False 104 ) 105 106 def is_auto_flush_reply_enabled(self) -> bool: 107 """ 108 is_auto_flush_reply_enabled: Returns True if the client is configured to auto-flush for listen. 109 """ 110 auto_flush_reply_delta = float(self.options.get("auto_flush_reply_delta", 0)) 111 return ( 112 isinstance(auto_flush_reply_delta, numbers.Number) 113 and auto_flush_reply_delta > 0 114 ) 115 116 def is_auto_flush_speak_enabled(self) -> bool: 117 """ 118 is_auto_flush_speak_enabled: Returns True if the client is configured to auto-flush for speak. 119 """ 120 auto_flush_speak_delta = float(self.options.get("auto_flush_speak_delta", 0)) 121 return ( 122 isinstance(auto_flush_speak_delta, numbers.Number) 123 and auto_flush_speak_delta > 0 124 ) 125 126 def is_inspecting_listen(self) -> bool: 127 """ 128 is_inspecting_listen: Returns True if the client is inspecting listen messages. 129 """ 130 return self._inspect_listen 131 132 def is_inspecting_speak(self) -> bool: 133 """ 134 is_inspecting_speak: Returns True if the client is inspecting speak messages. 135 """ 136 return self._inspect_speak 137 138 139class ClientOptionsFromEnv( 140 DeepgramClientOptions 141): # pylint: disable=too-many-branches, too-many-statements 142 """ 143 This class extends DeepgramClientOptions and will attempt to use environment variables first before defaults. 144 """ 145 146 _logger: verboselogs.VerboseLogger 147 148 def __init__( 149 self, 150 api_key: str = "", 151 url: str = "", 152 verbose: int = verboselogs.WARNING, 153 headers: Optional[Dict] = None, 154 options: Optional[Dict] = None, 155 ): # pylint: disable=too-many-positional-arguments 156 self._logger = verboselogs.VerboseLogger(__name__) 157 self._logger.addHandler(logging.StreamHandler()) 158 self._logger.setLevel(verboselogs.WARNING) # temporary set for setup 159 160 if api_key is None: 161 api_key = "" 162 163 if api_key == "": 164 api_key = os.getenv("DEEPGRAM_API_KEY", "") 165 if api_key == "": 166 self._logger.critical("Deepgram API KEY is not set") 167 raise DeepgramApiKeyError("Deepgram API KEY is not set") 168 169 if url == "": 170 url = os.getenv("DEEPGRAM_HOST", "api.deepgram.com") 171 self._logger.notice(f"Deepgram host is set to {url}") 172 173 if verbose == verboselogs.WARNING: 174 _loglevel = os.getenv("DEEPGRAM_LOGGING", "") 175 if _loglevel != "": 176 verbose = int(_loglevel) 177 if isinstance(verbose, str): 178 match verbose: 179 case "NOTSET": 180 self._logger.notice("Logging level is set to NOTSET") 181 verbose = verboselogs.NOTSET 182 case "SPAM": 183 self._logger.notice("Logging level is set to SPAM") 184 verbose = verboselogs.SPAM 185 case "DEBUG": 186 self._logger.notice("Logging level is set to DEBUG") 187 verbose = verboselogs.DEBUG 188 case "VERBOSE": 189 self._logger.notice("Logging level is set to VERBOSE") 190 verbose = verboselogs.VERBOSE 191 case "NOTICE": 192 self._logger.notice("Logging level is set to NOTICE") 193 verbose = verboselogs.NOTICE 194 case "WARNING": 195 self._logger.notice("Logging level is set to WARNING") 196 verbose = verboselogs.WARNING 197 case "SUCCESS": 198 self._logger.notice("Logging level is set to SUCCESS") 199 verbose = verboselogs.SUCCESS 200 case "ERROR": 201 self._logger.notice("Logging level is set to ERROR") 202 verbose = verboselogs.ERROR 203 case "CRITICAL": 204 self._logger.notice("Logging level is set to CRITICAL") 205 verbose = verboselogs.CRITICAL 206 case _: 207 self._logger.notice("Logging level is set to WARNING") 208 verbose = verboselogs.WARNING 209 self._logger.notice(f"Logging level is set to {verbose}") 210 211 if headers is None: 212 headers = {} 213 for x in range(0, 20): 214 header = os.getenv(f"DEEPGRAM_HEADER_{x}", None) 215 if header is not None: 216 headers[header] = os.getenv(f"DEEPGRAM_HEADER_VALUE_{x}", None) 217 self._logger.debug( 218 "Deepgram header %s is set with value %s", 219 header, 220 headers[header], 221 ) 222 else: 223 break 224 if len(headers) == 0: 225 self._logger.notice("Deepgram headers are not set") 226 headers = None 227 228 if options is None: 229 options = {} 230 for x in range(0, 20): 231 param = os.getenv(f"DEEPGRAM_PARAM_{x}", None) 232 if param is not None: 233 options[param] = os.getenv(f"DEEPGRAM_PARAM_VALUE_{x}", None) 234 self._logger.debug( 235 "Deepgram option %s is set with value %s", param, options[param] 236 ) 237 else: 238 break 239 if len(options) == 0: 240 self._logger.notice("Deepgram options are not set") 241 options = None 242 243 super().__init__( 244 api_key=api_key, url=url, verbose=verbose, headers=headers, options=options 245 )
class
DeepgramClientOptions:
18class DeepgramClientOptions: # pylint: disable=too-many-instance-attributes 19 """ 20 Represents options for configuring a Deepgram client. 21 22 This class allows you to customize various options for interacting with the Deepgram API. 23 24 Attributes: 25 api_key: (Optional) A Deepgram API key used for authentication. Default uses the `DEEPGRAM_API_KEY` environment variable. 26 url: (Optional) The URL used to interact with production, On-prem, and other Deepgram environments. Defaults to `api.deepgram.com`. 27 verbose: (Optional) The logging level for the client. Defaults to `verboselogs.WARNING`. 28 headers: (Optional) Headers for initializing the client. 29 options: (Optional) Additional options for initializing the client. 30 """ 31 32 _logger: verboselogs.VerboseLogger 33 _inspect_listen: bool = False 34 _inspect_speak: bool = False 35 36 def __init__( 37 self, 38 api_key: str = "", 39 url: str = "", 40 verbose: int = verboselogs.WARNING, 41 headers: Optional[Dict] = None, 42 options: Optional[Dict] = None, 43 ): # pylint: disable=too-many-positional-arguments 44 self._logger = verboselogs.VerboseLogger(__name__) 45 self._logger.addHandler(logging.StreamHandler()) 46 47 if api_key is None: 48 api_key = "" 49 50 self.verbose = verbose 51 self.api_key = api_key 52 53 if headers is None: 54 headers = {} 55 self._update_headers(headers=headers) 56 57 if len(url) == 0: 58 url = "api.deepgram.com" 59 self.url = self._get_url(url) 60 61 if options is None: 62 options = {} 63 self.options = options 64 65 if self.is_auto_flush_reply_enabled(): 66 self._inspect_listen = True 67 if self.is_auto_flush_speak_enabled(): 68 self._inspect_speak = True 69 70 def set_apikey(self, api_key: str): 71 """ 72 set_apikey: Sets the API key for the client. 73 74 Args: 75 api_key: The Deepgram API key used for authentication. 76 """ 77 self.api_key = api_key 78 self._update_headers() 79 80 def _get_url(self, url) -> str: 81 if not re.match(r"^https?://", url, re.IGNORECASE): 82 url = "https://" + url 83 return url.strip("/") 84 85 def _update_headers(self, headers: Optional[Dict] = None): 86 self.headers = {} 87 self.headers["Accept"] = "application/json" 88 if self.api_key: 89 self.headers["Authorization"] = f"Token {self.api_key}" 90 elif "Authorization" in self.headers: 91 del self.headers["Authorization"] 92 self.headers[ 93 "User-Agent" 94 ] = f"@deepgram/sdk/{__version__} python/{sys.version_info[1]}.{sys.version_info[2]}" 95 # Overwrite / add any headers that were passed in 96 if headers: 97 self.headers.update(headers) 98 99 def is_keep_alive_enabled(self) -> bool: 100 """ 101 is_keep_alive_enabled: Returns True if the client is configured to keep the connection alive. 102 """ 103 return self.options.get("keepalive", False) or self.options.get( 104 "keep_alive", False 105 ) 106 107 def is_auto_flush_reply_enabled(self) -> bool: 108 """ 109 is_auto_flush_reply_enabled: Returns True if the client is configured to auto-flush for listen. 110 """ 111 auto_flush_reply_delta = float(self.options.get("auto_flush_reply_delta", 0)) 112 return ( 113 isinstance(auto_flush_reply_delta, numbers.Number) 114 and auto_flush_reply_delta > 0 115 ) 116 117 def is_auto_flush_speak_enabled(self) -> bool: 118 """ 119 is_auto_flush_speak_enabled: Returns True if the client is configured to auto-flush for speak. 120 """ 121 auto_flush_speak_delta = float(self.options.get("auto_flush_speak_delta", 0)) 122 return ( 123 isinstance(auto_flush_speak_delta, numbers.Number) 124 and auto_flush_speak_delta > 0 125 ) 126 127 def is_inspecting_listen(self) -> bool: 128 """ 129 is_inspecting_listen: Returns True if the client is inspecting listen messages. 130 """ 131 return self._inspect_listen 132 133 def is_inspecting_speak(self) -> bool: 134 """ 135 is_inspecting_speak: Returns True if the client is inspecting speak messages. 136 """ 137 return self._inspect_speak
Represents options for configuring a Deepgram client.
This class allows you to customize various options for interacting with the Deepgram API.
Attributes:
api_key: (Optional) A Deepgram API key used for authentication. Default uses the DEEPGRAM_API_KEY environment variable.
url: (Optional) The URL used to interact with production, On-prem, and other Deepgram environments. Defaults to api.deepgram.com.
verbose: (Optional) The logging level for the client. Defaults to verboselogs.WARNING.
headers: (Optional) Headers for initializing the client.
options: (Optional) Additional options for initializing the client.
DeepgramClientOptions( api_key: str = '', url: str = '', verbose: int = 30, headers: Optional[Dict] = None, options: Optional[Dict] = None)
36 def __init__( 37 self, 38 api_key: str = "", 39 url: str = "", 40 verbose: int = verboselogs.WARNING, 41 headers: Optional[Dict] = None, 42 options: Optional[Dict] = None, 43 ): # pylint: disable=too-many-positional-arguments 44 self._logger = verboselogs.VerboseLogger(__name__) 45 self._logger.addHandler(logging.StreamHandler()) 46 47 if api_key is None: 48 api_key = "" 49 50 self.verbose = verbose 51 self.api_key = api_key 52 53 if headers is None: 54 headers = {} 55 self._update_headers(headers=headers) 56 57 if len(url) == 0: 58 url = "api.deepgram.com" 59 self.url = self._get_url(url) 60 61 if options is None: 62 options = {} 63 self.options = options 64 65 if self.is_auto_flush_reply_enabled(): 66 self._inspect_listen = True 67 if self.is_auto_flush_speak_enabled(): 68 self._inspect_speak = True
def
set_apikey(self, api_key: str):
70 def set_apikey(self, api_key: str): 71 """ 72 set_apikey: Sets the API key for the client. 73 74 Args: 75 api_key: The Deepgram API key used for authentication. 76 """ 77 self.api_key = api_key 78 self._update_headers()
set_apikey: Sets the API key for the client.
Args: api_key: The Deepgram API key used for authentication.
def
is_keep_alive_enabled(self) -> bool:
99 def is_keep_alive_enabled(self) -> bool: 100 """ 101 is_keep_alive_enabled: Returns True if the client is configured to keep the connection alive. 102 """ 103 return self.options.get("keepalive", False) or self.options.get( 104 "keep_alive", False 105 )
is_keep_alive_enabled: Returns True if the client is configured to keep the connection alive.
def
is_auto_flush_reply_enabled(self) -> bool:
107 def is_auto_flush_reply_enabled(self) -> bool: 108 """ 109 is_auto_flush_reply_enabled: Returns True if the client is configured to auto-flush for listen. 110 """ 111 auto_flush_reply_delta = float(self.options.get("auto_flush_reply_delta", 0)) 112 return ( 113 isinstance(auto_flush_reply_delta, numbers.Number) 114 and auto_flush_reply_delta > 0 115 )
is_auto_flush_reply_enabled: Returns True if the client is configured to auto-flush for listen.
def
is_auto_flush_speak_enabled(self) -> bool:
117 def is_auto_flush_speak_enabled(self) -> bool: 118 """ 119 is_auto_flush_speak_enabled: Returns True if the client is configured to auto-flush for speak. 120 """ 121 auto_flush_speak_delta = float(self.options.get("auto_flush_speak_delta", 0)) 122 return ( 123 isinstance(auto_flush_speak_delta, numbers.Number) 124 and auto_flush_speak_delta > 0 125 )
is_auto_flush_speak_enabled: Returns True if the client is configured to auto-flush for speak.
140class ClientOptionsFromEnv( 141 DeepgramClientOptions 142): # pylint: disable=too-many-branches, too-many-statements 143 """ 144 This class extends DeepgramClientOptions and will attempt to use environment variables first before defaults. 145 """ 146 147 _logger: verboselogs.VerboseLogger 148 149 def __init__( 150 self, 151 api_key: str = "", 152 url: str = "", 153 verbose: int = verboselogs.WARNING, 154 headers: Optional[Dict] = None, 155 options: Optional[Dict] = None, 156 ): # pylint: disable=too-many-positional-arguments 157 self._logger = verboselogs.VerboseLogger(__name__) 158 self._logger.addHandler(logging.StreamHandler()) 159 self._logger.setLevel(verboselogs.WARNING) # temporary set for setup 160 161 if api_key is None: 162 api_key = "" 163 164 if api_key == "": 165 api_key = os.getenv("DEEPGRAM_API_KEY", "") 166 if api_key == "": 167 self._logger.critical("Deepgram API KEY is not set") 168 raise DeepgramApiKeyError("Deepgram API KEY is not set") 169 170 if url == "": 171 url = os.getenv("DEEPGRAM_HOST", "api.deepgram.com") 172 self._logger.notice(f"Deepgram host is set to {url}") 173 174 if verbose == verboselogs.WARNING: 175 _loglevel = os.getenv("DEEPGRAM_LOGGING", "") 176 if _loglevel != "": 177 verbose = int(_loglevel) 178 if isinstance(verbose, str): 179 match verbose: 180 case "NOTSET": 181 self._logger.notice("Logging level is set to NOTSET") 182 verbose = verboselogs.NOTSET 183 case "SPAM": 184 self._logger.notice("Logging level is set to SPAM") 185 verbose = verboselogs.SPAM 186 case "DEBUG": 187 self._logger.notice("Logging level is set to DEBUG") 188 verbose = verboselogs.DEBUG 189 case "VERBOSE": 190 self._logger.notice("Logging level is set to VERBOSE") 191 verbose = verboselogs.VERBOSE 192 case "NOTICE": 193 self._logger.notice("Logging level is set to NOTICE") 194 verbose = verboselogs.NOTICE 195 case "WARNING": 196 self._logger.notice("Logging level is set to WARNING") 197 verbose = verboselogs.WARNING 198 case "SUCCESS": 199 self._logger.notice("Logging level is set to SUCCESS") 200 verbose = verboselogs.SUCCESS 201 case "ERROR": 202 self._logger.notice("Logging level is set to ERROR") 203 verbose = verboselogs.ERROR 204 case "CRITICAL": 205 self._logger.notice("Logging level is set to CRITICAL") 206 verbose = verboselogs.CRITICAL 207 case _: 208 self._logger.notice("Logging level is set to WARNING") 209 verbose = verboselogs.WARNING 210 self._logger.notice(f"Logging level is set to {verbose}") 211 212 if headers is None: 213 headers = {} 214 for x in range(0, 20): 215 header = os.getenv(f"DEEPGRAM_HEADER_{x}", None) 216 if header is not None: 217 headers[header] = os.getenv(f"DEEPGRAM_HEADER_VALUE_{x}", None) 218 self._logger.debug( 219 "Deepgram header %s is set with value %s", 220 header, 221 headers[header], 222 ) 223 else: 224 break 225 if len(headers) == 0: 226 self._logger.notice("Deepgram headers are not set") 227 headers = None 228 229 if options is None: 230 options = {} 231 for x in range(0, 20): 232 param = os.getenv(f"DEEPGRAM_PARAM_{x}", None) 233 if param is not None: 234 options[param] = os.getenv(f"DEEPGRAM_PARAM_VALUE_{x}", None) 235 self._logger.debug( 236 "Deepgram option %s is set with value %s", param, options[param] 237 ) 238 else: 239 break 240 if len(options) == 0: 241 self._logger.notice("Deepgram options are not set") 242 options = None 243 244 super().__init__( 245 api_key=api_key, url=url, verbose=verbose, headers=headers, options=options 246 )
This class extends DeepgramClientOptions and will attempt to use environment variables first before defaults.
ClientOptionsFromEnv( api_key: str = '', url: str = '', verbose: int = 30, headers: Optional[Dict] = None, options: Optional[Dict] = None)
149 def __init__( 150 self, 151 api_key: str = "", 152 url: str = "", 153 verbose: int = verboselogs.WARNING, 154 headers: Optional[Dict] = None, 155 options: Optional[Dict] = None, 156 ): # pylint: disable=too-many-positional-arguments 157 self._logger = verboselogs.VerboseLogger(__name__) 158 self._logger.addHandler(logging.StreamHandler()) 159 self._logger.setLevel(verboselogs.WARNING) # temporary set for setup 160 161 if api_key is None: 162 api_key = "" 163 164 if api_key == "": 165 api_key = os.getenv("DEEPGRAM_API_KEY", "") 166 if api_key == "": 167 self._logger.critical("Deepgram API KEY is not set") 168 raise DeepgramApiKeyError("Deepgram API KEY is not set") 169 170 if url == "": 171 url = os.getenv("DEEPGRAM_HOST", "api.deepgram.com") 172 self._logger.notice(f"Deepgram host is set to {url}") 173 174 if verbose == verboselogs.WARNING: 175 _loglevel = os.getenv("DEEPGRAM_LOGGING", "") 176 if _loglevel != "": 177 verbose = int(_loglevel) 178 if isinstance(verbose, str): 179 match verbose: 180 case "NOTSET": 181 self._logger.notice("Logging level is set to NOTSET") 182 verbose = verboselogs.NOTSET 183 case "SPAM": 184 self._logger.notice("Logging level is set to SPAM") 185 verbose = verboselogs.SPAM 186 case "DEBUG": 187 self._logger.notice("Logging level is set to DEBUG") 188 verbose = verboselogs.DEBUG 189 case "VERBOSE": 190 self._logger.notice("Logging level is set to VERBOSE") 191 verbose = verboselogs.VERBOSE 192 case "NOTICE": 193 self._logger.notice("Logging level is set to NOTICE") 194 verbose = verboselogs.NOTICE 195 case "WARNING": 196 self._logger.notice("Logging level is set to WARNING") 197 verbose = verboselogs.WARNING 198 case "SUCCESS": 199 self._logger.notice("Logging level is set to SUCCESS") 200 verbose = verboselogs.SUCCESS 201 case "ERROR": 202 self._logger.notice("Logging level is set to ERROR") 203 verbose = verboselogs.ERROR 204 case "CRITICAL": 205 self._logger.notice("Logging level is set to CRITICAL") 206 verbose = verboselogs.CRITICAL 207 case _: 208 self._logger.notice("Logging level is set to WARNING") 209 verbose = verboselogs.WARNING 210 self._logger.notice(f"Logging level is set to {verbose}") 211 212 if headers is None: 213 headers = {} 214 for x in range(0, 20): 215 header = os.getenv(f"DEEPGRAM_HEADER_{x}", None) 216 if header is not None: 217 headers[header] = os.getenv(f"DEEPGRAM_HEADER_VALUE_{x}", None) 218 self._logger.debug( 219 "Deepgram header %s is set with value %s", 220 header, 221 headers[header], 222 ) 223 else: 224 break 225 if len(headers) == 0: 226 self._logger.notice("Deepgram headers are not set") 227 headers = None 228 229 if options is None: 230 options = {} 231 for x in range(0, 20): 232 param = os.getenv(f"DEEPGRAM_PARAM_{x}", None) 233 if param is not None: 234 options[param] = os.getenv(f"DEEPGRAM_PARAM_VALUE_{x}", None) 235 self._logger.debug( 236 "Deepgram option %s is set with value %s", param, options[param] 237 ) 238 else: 239 break 240 if len(options) == 0: 241 self._logger.notice("Deepgram options are not set") 242 options = None 243 244 super().__init__( 245 api_key=api_key, url=url, verbose=verbose, headers=headers, options=options 246 )