Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion deepgram/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from typing import Optional
from importlib import import_module
import logging, verboselogs
import os

from .clients.listen import ListenClient, PreRecordedClient
from .clients.manage.client import ManageClient
Expand Down Expand Up @@ -39,7 +40,8 @@ def __init__(self, api_key: str, config: Optional[DeepgramClientOptions] = None)
self.logger.addHandler(logging.StreamHandler())

if not api_key:
raise DeepgramApiKeyError("Deepgram API key is required")
# Default to `None` for on-prem instances where an API key is not required
api_key = os.getenv("DEEPGRAM_API_KEY", None)

self.api_key = api_key
if config is None: # Use default configuration
Expand Down
28 changes: 15 additions & 13 deletions deepgram/options.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
# SPDX-License-Identifier: MIT

import logging, verboselogs
from typing import Dict
from typing import Dict, Optional
import re


Expand Down Expand Up @@ -31,15 +31,7 @@ def __init__(
):
self.verbose = verbose
self.api_key = api_key
if headers is None:
self.headers = {
"Accept": "application/json",
"Authorization": f"Token {self.api_key}",
}
else:
self.headers.update(
{"Accept": "application/json", "Authorization": f"Token {self.api_key}"}
)
self._update_headers(headers=headers)
if len(url) == 0:
url = "api.deepgram.com"
self.url = self._get_url(url)
Expand All @@ -49,11 +41,21 @@ def __init__(

def set_apikey(self, api_key: str):
self.api_key = api_key
self.headers.update(
{"Accept": "application/json", "Authorization": f"Token {self.api_key}"}
)
self._update_headers()

def _get_url(self, url):
if not re.match(r"^https?://", url, re.IGNORECASE):
url = "https://" + url
return url.strip("/")

def _update_headers(self, headers: Optional[Dict[str, str]] = None):
if not hasattr(self, "headers") or self.headers is None:
self.headers = {}
self.headers["Accept"] = "application/json"
if self.api_key:
self.headers["Authorization"] = f"Token {self.api_key}"
elif "Authorization" in self.headers:
del self.headers["Authorization"]
# Overwrite / add any headers that were passed in
if headers:
self.headers.update(headers)