Implement EnvVar for Containerized Images

Author: davidvonthenen

deepgram/client.py

@@ -77,7 +77,7 @@
 from .clients.onprem.client import OnPremClient
 from .clients.onprem.v1.async_client import AsyncOnPremClient
 
-from .options import DeepgramClientOptions
+from .options import DeepgramClientOptions, ClientOptionsFromEnv
 from .errors import DeepgramApiKeyError, DeepgramModuleError
 
 
@@ -128,8 +128,16 @@ class DeepgramClient:
     """
 
     def __init__(
-        self, api_key: str = "", config: Optional[DeepgramClientOptions] = None
+        self,
+        api_key: str = "",
+        config: Optional[DeepgramClientOptions] = None,
     ):
+        verboselogs.install()
+        self.logger = logging.getLogger(__name__)
+        self.logger.addHandler(logging.StreamHandler())
+
+        if config is not None:
+            api_key = config.api_key
         if not api_key:
             # Default to `None` for on-prem instances where an API key is not required
             api_key = os.getenv("DEEPGRAM_API_KEY", None)

deepgram/options.py

@@ -38,9 +38,11 @@ def __init__(
         self.verbose = verbose
         self.api_key = api_key
         self._update_headers(headers=headers)
+
         if len(url) == 0:
             url = "api.deepgram.com"
         self.url = self._get_url(url)
+
         if options is None:
             options = dict()
         self.options = options
@@ -70,18 +72,93 @@ def _update_headers(self, headers: Optional[Dict[str, str]] = None):
 class ClientOptionsFromEnv(DeepgramClientOptions):
     def __init__(
         self,
+        api_key: str = "",
+        url: str = "",
         verbose: int = logging.WARNING,
         headers: Dict[str, str] = None,
         options: Dict[str, str] = None,
     ):
-        apiKey = os.getenv("DEEPGRAM_API_KEY", None)
-        if apiKey is None:
-            raise DeepgramApiKeyError("Deepgram API KEY is not set")
+        verboselogs.install()
+        self.logger = logging.getLogger(__name__)
+        self.logger.addHandler(logging.StreamHandler())
+        self.logger.setLevel(logging.WARNING)  # temporary set for setup
 
-        url = os.getenv("DEEPGRAM_URL", None)
-        if url is None:
-            url = "api.deepgram.com"
+        if api_key == "":
+            api_key = os.getenv("DEEPGRAM_API_KEY", None)
+            if api_key is None:
+                self.logger.critical("Deepgram API KEY is not set")
+                raise DeepgramApiKeyError("Deepgram API KEY is not set")
+
+        if url == "":
+            url = os.getenv("DEEPGRAM_HOST", "api.deepgram.com")
+            self.logger.notice(f"Deepgram host is set to {url}")
+
+        if verbose == logging.WARNING:
+            verbose = os.getenv("DEEPGRAM_LOGGING", logging.WARNING)
+            if type(verbose) != int:
+                match verbose:
+                    case "NOTSET":
+                        self.logger.notice("Logging level is set to NOTSET")
+                        verbose = logging.NOTSET
+                    case "SPAM":
+                        self.logger.notice("Logging level is set to SPAM")
+                        verbose = logging.SPAM
+                    case "DEBUG":
+                        self.logger.notice("Logging level is set to DEBUG")
+                        verbose = logging.DEBUG
+                    case "VERBOSE":
+                        self.logger.notice("Logging level is set to VERBOSE")
+                        verbose = logging.VERBOSE
+                    case "NOTICE":
+                        self.logger.notice("Logging level is set to NOTICE")
+                        verbose = logging.NOTICE
+                    case "WARNING":
+                        self.logger.notice("Logging level is set to WARNING")
+                        verbose = logging.WARNING
+                    case "SUCCESS":
+                        self.logger.notice("Logging level is set to SUCCESS")
+                        verbose = logging.SUCCESS
+                    case "ERROR":
+                        self.logger.notice("Logging level is set to ERROR")
+                        verbose = logging.ERROR
+                    case "CRITICAL":
+                        self.logger.notice("Logging level is set to CRITICAL")
+                        verbose = logging.CRITICAL
+                    case _:
+                        self.logger.notice("Logging level is set to WARNING")
+                        verbose = logging.WARNING
+        self.logger.notice(f"Logging level is set to {verbose}")
+
+        if headers is None:
+            headers = dict()
+            for x in range(0, 20):
+                header = os.getenv(f"DEEPGRAM_HEADER_{x}", None)
+                if header is not None:
+                    headers[header] = os.getenv(f"DEEPGRAM_HEADER_VALUE_{x}", None)
+                    self.logger.debug(
+                        f"Deepgram header {header} is set with value {headers[header]}"
+                    )
+                else:
+                    break
+            if len(headers) == 0:
+                self.logger.notice("Deepgram headers are not set")
+                headers = None
+
+        if options is None:
+            options = dict()
+            for x in range(0, 20):
+                param = os.getenv(f"DEEPGRAM_PARAM_{x}", None)
+                if param is not None:
+                    options[param] = os.getenv(f"DEEPGRAM_PARAM_VALUE_{x}", None)
+                    self.logger.debug(
+                        f"Deepgram option {param} is set with value {options[param]}"
+                    )
+                else:
+                    break
+            if len(options) == 0:
+                self.logger.notice("Deepgram options are not set")
+                options = None
 
         super().__init__(
-            api_key=apiKey, url=url, verbose=verbose, headers=headers, options=options
+            api_key=api_key, url=url, verbose=verbose, headers=headers, options=options
         )

examples/prerecorded/file/main.py

@@ -8,8 +8,8 @@
 from datetime import datetime, timedelta
 
 from deepgram import (
-    DeepgramClientOptions,
     DeepgramClient,
+    DeepgramClientOptions,
     PrerecordedOptions,
     FileSource,
 )
@@ -22,11 +22,11 @@
 def main():
     try:
         # STEP 1 Create a Deepgram client using the API key in the environment variables
-        config: DeepgramClientOptions = DeepgramClientOptions(
+        config = DeepgramClientOptions(
             verbose=logging.SPAM,
         )
 
-        deepgram: DeepgramClient = DeepgramClient("", config)
+        deepgram = DeepgramClient("", config)
 
         # STEP 2 Call the transcribe_file method on the prerecorded class
         with open(AUDIO_FILE, "rb") as file:

examples/prerecorded/url/main.py

@@ -6,7 +6,11 @@
 from dotenv import load_dotenv
 import logging, verboselogs
 
-from deepgram import DeepgramClient, DeepgramClientOptions, PrerecordedOptions
+from deepgram import (
+    DeepgramClient,
+    ClientOptionsFromEnv,
+    PrerecordedOptions,
+)
 
 load_dotenv()
 
@@ -18,7 +22,7 @@
 def main():
     try:
         # STEP 1 Create a Deepgram client using the API key from environment variables
-        deepgram = DeepgramClient()
+        deepgram = DeepgramClient("", ClientOptionsFromEnv())
 
         # STEP 2 Call the transcribe_url method on the prerecorded class
         options = PrerecordedOptions(

examples/streaming/async_http/main.py

@@ -11,6 +11,8 @@
 
 load_dotenv()
 
+API_KEY = os.getenv("DG_API_KEY")
+
 options = LiveOptions(
     model="nova",
     interim_results=False,
@@ -22,7 +24,7 @@
 
 
 async def main():
-    deepgram = DeepgramClient()
+    deepgram = DeepgramClient(API_KEY)
 
     # Create a websocket connection to Deepgram
     try: