refactor(passthrough): share the Deepgram WebSocket route body

The listen route's body (provider credential, server-side
DEEPGRAM_API_BASE, early refusal, price check, relay call) becomes
_relay_deepgram_websocket, which takes a small _DeepgramWebsocketProduct
record naming what differs per product: the target URL builder, the price
check, the registry key and an optional refusal. The listen auth
dependency and the listen route keep their names, paths and order

In common_utils, the target URL builder and the upstream model lookup take
the path and default model as arguments, and the exact registry lookup
takes the registry key, so a second product can reuse them

No behavior changes. The listen tests pass unchanged

Refs LIT-10070

litellm/llms/deepgram/common_utils.py

@@ -34,8 +34,12 @@ class DeepgramException(BaseLLMException):
     pass
 
 
+def _requested_model(query_string: str, default_model: str) -> str:
+    return httpx.QueryParams(query_string).get("model") or default_model
+
+
 def deepgram_listen_requested_model(query_string: str) -> str:
-    return httpx.QueryParams(query_string).get("model") or DEEPGRAM_LISTEN_DEFAULT_MODEL
+    return _requested_model(query_string, DEEPGRAM_LISTEN_DEFAULT_MODEL)
 
 
 def _first_occurrences(query_string: str) -> httpx.QueryParams:
@@ -50,21 +54,29 @@ def _first_occurrences(query_string: str) -> httpx.QueryParams:
     )
 
 
-def deepgram_listen_websocket_target(api_base: str | None, query_string: str) -> str:
-    listen_url: Final = httpx.URL(join_api_base(api_base or DEEPGRAM_DEFAULT_API_BASE, "listen"))
-    websocket_url: Final = listen_url.copy_with(scheme=_WEBSOCKET_SCHEMES.get(listen_url.scheme, listen_url.scheme))
+def _deepgram_websocket_target(api_base: str | None, path: str, query_string: str, default_model: str) -> str:
+    http_url: Final = httpx.URL(join_api_base(api_base or DEEPGRAM_DEFAULT_API_BASE, path))
+    websocket_url: Final = http_url.copy_with(scheme=_WEBSOCKET_SCHEMES.get(http_url.scheme, http_url.scheme))
     params: Final = _first_occurrences(query_string)
-    query: Final = params if params.get("model") else params.remove("model").add("model", DEEPGRAM_LISTEN_DEFAULT_MODEL)
+    query: Final = params if params.get("model") else params.remove("model").add("model", default_model)
     return f"{websocket_url}?{query}"
 
 
+def deepgram_listen_websocket_target(api_base: str | None, query_string: str) -> str:
+    return _deepgram_websocket_target(api_base, "listen", query_string, DEEPGRAM_LISTEN_DEFAULT_MODEL)
+
+
 def deepgram_listen_callback_params(query_string: str) -> tuple[str, ...]:
     return tuple(sorted(DEEPGRAM_LISTEN_CALLBACK_PARAMS.intersection(httpx.QueryParams(query_string).keys())))
 
 
-def deepgram_listen_model(upstream_url: str) -> str:
+def _upstream_model(upstream_url: str, default_model: str) -> str:
     models: Final = parse_qs(urlparse(upstream_url).query).get("model")
-    return models[0] if models else DEEPGRAM_LISTEN_DEFAULT_MODEL
+    return models[0] if models else default_model
+
+
+def deepgram_listen_model(upstream_url: str) -> str:
+    return _upstream_model(upstream_url, DEEPGRAM_LISTEN_DEFAULT_MODEL)
 
 
 def _param_enabled(values: Sequence[str]) -> bool:
@@ -89,7 +101,10 @@ def deepgram_listen_registry_key(upstream_url: str) -> str:
 def deepgram_listen_is_priced(upstream_url: str) -> bool:
     """Only an exact registry hit counts: the cost calculator resolves a missing ``streaming/<model>`` row to the
     pre-recorded ``<model>`` row, which is not the rate Deepgram bills a WebSocket session at."""
-    registry_key: Final = deepgram_listen_registry_key(upstream_url)
+    return _has_exact_registry_entry(deepgram_listen_registry_key(upstream_url))
+
+
+def _has_exact_registry_entry(registry_key: str) -> bool:
     try:
         model_info: Final = litellm.get_model_info(model=registry_key, custom_llm_provider=LlmProviders.DEEPGRAM.value)
     except Exception:

litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py

@@ -3469,12 +3469,21 @@ async def deepgram_listen_user_api_key_auth(websocket: WebSocket) -> UserAPIKeyA
     )
 
 
-@router.websocket("/deepgram/v1/listen")
-@router.websocket("/deepgram/listen")
-async def deepgram_listen_websocket_route(
+def _deepgram_listen_refusal(target: str) -> str | None:
+    callback_params: Final = deepgram_listen_callback_params(httpx.URL(target).query.decode())
+    if callback_params:
+        return _DEEPGRAM_WS_CALLBACK_REASON.format(params=", ".join(callback_params))
+    if deepgram_listen_is_priced(target):
+        return None
+    return _DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_listen_registry_key(target))
+
+
+async def _relay_deepgram_websocket(
     websocket: WebSocket,
-    user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(deepgram_listen_user_api_key_auth)],
-    relay: Annotated[_WebsocketRelay, Depends(_websocket_relay)],
+    user_api_key_dict: UserAPIKeyAuth,
+    relay: _WebsocketRelay,
+    target: str,
+    refusal: str | None,
 ) -> None:
     deepgram_api_key: Final = passthrough_endpoint_router.get_credentials(
         custom_llm_provider=litellm.LlmProviders.DEEPGRAM.value,
@@ -3485,23 +3494,8 @@ async def deepgram_listen_websocket_route(
         return
 
     await websocket.accept(subprotocol=_negotiated_websocket_subprotocol(websocket))
-    callback_params: Final = deepgram_listen_callback_params(websocket.url.query)
-    if callback_params:
-        await websocket.close(
-            code=1008,
-            reason=_DEEPGRAM_WS_CALLBACK_REASON.format(params=", ".join(callback_params)),
-        )
-        return
-
-    target: Final = deepgram_listen_websocket_target(
-        api_base=get_secret_str("DEEPGRAM_API_BASE"),
-        query_string=websocket.url.query,
-    )
-    if not deepgram_listen_is_priced(target):
-        await websocket.close(
-            code=1008,
-            reason=_DEEPGRAM_WS_UNPRICED_REASON.format(registry_key=deepgram_listen_registry_key(target)),
-        )
+    if refusal is not None:
+        await websocket.close(code=1008, reason=refusal)
         return
 
     await relay(
@@ -3515,6 +3509,17 @@ async def deepgram_listen_websocket_route(
     )
 
 
+@router.websocket("/deepgram/v1/listen")
+@router.websocket("/deepgram/listen")
+async def deepgram_listen_websocket_route(
+    websocket: WebSocket,
+    user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(deepgram_listen_user_api_key_auth)],
+    relay: Annotated[_WebsocketRelay, Depends(_websocket_relay)],
+) -> None:
+    target: Final = deepgram_listen_websocket_target(get_secret_str("DEEPGRAM_API_BASE"), websocket.url.query)
+    await _relay_deepgram_websocket(websocket, user_api_key_dict, relay, target, _deepgram_listen_refusal(target))
+
+
 class BaseOpenAIPassThroughHandler:
     @staticmethod
     async def base_openai_pass_through_handler(