|
| 1 | +""" |
| 2 | +Custom client entrypoints that extend the generated BaseClient/AsyncBaseClient. |
| 3 | +
|
| 4 | +Adds support for: |
| 5 | +- `access_token` as an alternative to `api_key` with the following rules: |
| 6 | + - If `access_token` is provided, it takes precedence and sets `Authorization: bearer <token>` |
| 7 | + - When `access_token` is used, `api_key` is forced to "token" to satisfy the generator, |
| 8 | + but the Authorization header is overridden for all HTTP and WebSocket requests. |
| 9 | +- `session_id` as a header sent with every request and websocket connection: |
| 10 | + - If `session_id` is provided, it will be used; otherwise, a UUID is auto-generated |
| 11 | + - The session_id is sent as the `x-deepgram-session-id` header |
| 12 | +""" |
| 13 | + |
| 14 | +import types |
| 15 | +import uuid |
| 16 | +from typing import Any, Dict, Optional |
| 17 | + |
| 18 | +from .base_client import AsyncBaseClient, BaseClient |
| 19 | + |
| 20 | +from deepgram.core.client_wrapper import BaseClientWrapper |
| 21 | + |
| 22 | + |
| 23 | +def _apply_bearer_authorization_override(client_wrapper: BaseClientWrapper, bearer_token: str) -> None: |
| 24 | + """Override header providers to always use a Bearer authorization token. |
| 25 | +
|
| 26 | + This updates both: |
| 27 | + - client_wrapper.get_headers() used by WebSocket clients |
| 28 | + - client_wrapper.httpx_client.base_headers used by HTTP clients |
| 29 | + """ |
| 30 | + original_get_headers = client_wrapper.get_headers |
| 31 | + |
| 32 | + def _get_headers_with_bearer(_self: Any) -> Dict[str, str]: |
| 33 | + headers = original_get_headers() |
| 34 | + headers["Authorization"] = f"bearer {bearer_token}" |
| 35 | + return headers |
| 36 | + |
| 37 | + # Override on wrapper for WebSockets |
| 38 | + client_wrapper.get_headers = types.MethodType(_get_headers_with_bearer, client_wrapper) # type: ignore[method-assign] |
| 39 | + |
| 40 | + # Override on HTTP client for REST requests |
| 41 | + if hasattr(client_wrapper, "httpx_client") and hasattr(client_wrapper.httpx_client, "base_headers"): |
| 42 | + client_wrapper.httpx_client.base_headers = client_wrapper.get_headers |
| 43 | + |
| 44 | + |
| 45 | +class DeepgramClient(BaseClient): |
| 46 | + """ |
| 47 | + Custom Deepgram client that extends the generated BaseClient. |
| 48 | +
|
| 49 | + Supports: |
| 50 | + - `session_id`: Optional session identifier. If not provided, a UUID is auto-generated. |
| 51 | + Sent as `x-deepgram-session-id` header in all requests and websocket connections. |
| 52 | + - `access_token`: Alternative to `api_key`. If provided, uses Bearer token authentication. |
| 53 | + - `telemetry_opt_out`: Telemetry opt-out flag (maintained for backwards compatibility, no-op). |
| 54 | + - `telemetry_handler`: Telemetry handler (maintained for backwards compatibility, no-op). |
| 55 | + """ |
| 56 | + |
| 57 | + def __init__(self, *args, **kwargs) -> None: |
| 58 | + access_token: Optional[str] = kwargs.pop("access_token", None) |
| 59 | + session_id: Optional[str] = kwargs.pop("session_id", None) |
| 60 | + telemetry_opt_out: bool = bool(kwargs.pop("telemetry_opt_out", True)) |
| 61 | + telemetry_handler: Optional[Any] = kwargs.pop("telemetry_handler", None) |
| 62 | + |
| 63 | + # Use provided session_id or generate one |
| 64 | + final_session_id = session_id if session_id is not None else str(uuid.uuid4()) |
| 65 | + |
| 66 | + # Ensure headers object exists for pass-through custom headers |
| 67 | + headers: Optional[Dict[str, str]] = kwargs.get("headers") |
| 68 | + if headers is None: |
| 69 | + headers = {} |
| 70 | + kwargs["headers"] = headers |
| 71 | + |
| 72 | + # Ensure every request has a session identifier header |
| 73 | + headers["x-deepgram-session-id"] = final_session_id |
| 74 | + |
| 75 | + # Handle access_token: if provided, it takes precedence over api_key |
| 76 | + # The base client requires api_key, so we set a placeholder if needed |
| 77 | + # The Authorization header will be overridden to use Bearer token |
| 78 | + if access_token is not None: |
| 79 | + # Set a placeholder api_key if none provided (base client requires it) |
| 80 | + if kwargs.get("api_key") is None: |
| 81 | + kwargs["api_key"] = "token" |
| 82 | + |
| 83 | + super().__init__(*args, **kwargs) |
| 84 | + self.session_id = final_session_id |
| 85 | + |
| 86 | + # Override Authorization header to use Bearer token if access_token was provided |
| 87 | + if access_token is not None: |
| 88 | + _apply_bearer_authorization_override(self._client_wrapper, access_token) |
| 89 | + |
| 90 | + # Store telemetry handler for backwards compatibility (no-op, telemetry not implemented) |
| 91 | + self._telemetry_handler = None |
| 92 | + |
| 93 | + |
| 94 | +class AsyncDeepgramClient(AsyncBaseClient): |
| 95 | + """ |
| 96 | + Custom async Deepgram client that extends the generated AsyncBaseClient. |
| 97 | +
|
| 98 | + Supports: |
| 99 | + - `session_id`: Optional session identifier. If not provided, a UUID is auto-generated. |
| 100 | + Sent as `x-deepgram-session-id` header in all requests and websocket connections. |
| 101 | + - `access_token`: Alternative to `api_key`. If provided, uses Bearer token authentication. |
| 102 | + - `telemetry_opt_out`: Telemetry opt-out flag (maintained for backwards compatibility, no-op). |
| 103 | + - `telemetry_handler`: Telemetry handler (maintained for backwards compatibility, no-op). |
| 104 | + """ |
| 105 | + |
| 106 | + def __init__(self, *args, **kwargs) -> None: |
| 107 | + access_token: Optional[str] = kwargs.pop("access_token", None) |
| 108 | + session_id: Optional[str] = kwargs.pop("session_id", None) |
| 109 | + telemetry_opt_out: bool = bool(kwargs.pop("telemetry_opt_out", True)) |
| 110 | + telemetry_handler: Optional[Any] = kwargs.pop("telemetry_handler", None) |
| 111 | + |
| 112 | + # Use provided session_id or generate one |
| 113 | + final_session_id = session_id if session_id is not None else str(uuid.uuid4()) |
| 114 | + |
| 115 | + # Ensure headers object exists for pass-through custom headers |
| 116 | + headers: Optional[Dict[str, str]] = kwargs.get("headers") |
| 117 | + if headers is None: |
| 118 | + headers = {} |
| 119 | + kwargs["headers"] = headers |
| 120 | + |
| 121 | + # Ensure every request has a session identifier header |
| 122 | + headers["x-deepgram-session-id"] = final_session_id |
| 123 | + |
| 124 | + # Handle access_token: if provided, it takes precedence over api_key |
| 125 | + # The base client requires api_key, so we set a placeholder if needed |
| 126 | + # The Authorization header will be overridden to use Bearer token |
| 127 | + if access_token is not None: |
| 128 | + # Set a placeholder api_key if none provided (base client requires it) |
| 129 | + if kwargs.get("api_key") is None: |
| 130 | + kwargs["api_key"] = "token" |
| 131 | + |
| 132 | + super().__init__(*args, **kwargs) |
| 133 | + self.session_id = final_session_id |
| 134 | + |
| 135 | + # Override Authorization header to use Bearer token if access_token was provided |
| 136 | + if access_token is not None: |
| 137 | + _apply_bearer_authorization_override(self._client_wrapper, access_token) |
| 138 | + |
| 139 | + # Store telemetry handler for backwards compatibility (no-op, telemetry not implemented) |
| 140 | + self._telemetry_handler = None |
0 commit comments