mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-21 00:21:49 +00:00
Merge pull request #41515 from BerriAI/litellm_transcribe_passthrough
feat(proxy): add Amazon Transcribe pass-through with completion-time job pricing
This commit is contained in:
Yassin Kortam
GitHub
commit
f129e7e2d0
25 changed files with 2540 additions and 36 deletions
|
|
@ -85,6 +85,7 @@ GATEWAY_PATH_PREFIXES: tuple[str, ...] = (
|
|||
"/aws/",
|
||||
"/bedrock/",
|
||||
"/comprehendmedical",
|
||||
"/transcribe",
|
||||
"/cohere/",
|
||||
"/gemini/",
|
||||
"/gigachat/",
|
||||
|
|
|
|||
|
|
@ -66,7 +66,7 @@
|
|||
"/v1/video" "/v1/videos" "/video" "/videos" "/v1/search" "/search"
|
||||
"/v1/containers" "/containers" "/v1/evals" "/v1/memory" "/queue/chat"
|
||||
"/v1beta" "/interactions"
|
||||
"/anthropic" "/azure" "/azure_ai" "/aws" "/bedrock" "/comprehendmedical" "/cohere" "/gemini" "/google"
|
||||
"/anthropic" "/azure" "/azure_ai" "/aws" "/bedrock" "/comprehendmedical" "/transcribe" "/cohere" "/gemini" "/google"
|
||||
"/vertex_ai" "/vertex-ai" "/assemblyai" "/eu.assemblyai" "/langfuse" "/vllm"
|
||||
"/mistral" "/groq" "/voyage" "/cursor" "/milvus" "/openai_passthrough"
|
||||
"/toolset"
|
||||
|
|
|
|||
|
|
@ -1575,6 +1575,15 @@ PASS_THROUGH_HEADER_PREFIX: Final = "x-pass-"
|
|||
|
||||
BASE_MCP_ROUTE: Final = "/mcp"
|
||||
|
||||
TRANSCRIBE_JOB_POLLING_INTERVAL_SECONDS: Final = 10.0
|
||||
TRANSCRIBE_JOB_MAX_POLLING_ATTEMPTS: Final = 720 # 2 hours
|
||||
TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS: Final = 28800 # Amazon Transcribe quota: maximum audio file length
|
||||
TRANSCRIBE_MAX_MEDIA_BYTES: Final = 2 * 1024**3 # Amazon Transcribe quota: maximum audio file size
|
||||
TRANSCRIBE_MEDIA_DOWNLOAD_CONCURRENCY: Final = 1
|
||||
TRANSCRIBE_MEDIA_FETCH_ATTEMPTS: Final = 3
|
||||
TRANSCRIBE_MEDIA_LAST_MODIFIED_TOLERANCE_SECONDS: Final = 1.0 # S3 Last-Modified carries whole seconds only
|
||||
TRANSCRIBE_MEASURABLE_MEDIA_FORMATS: Final = frozenset({"flac", "mp3", "ogg", "wav"}) # what libsndfile can read
|
||||
|
||||
BATCH_STATUS_POLL_INTERVAL_SECONDS: Final = int(os.getenv("BATCH_STATUS_POLL_INTERVAL_SECONDS", 3600)) # 1 hour
|
||||
BATCH_STATUS_POLL_MAX_ATTEMPTS: Final = int(os.getenv("BATCH_STATUS_POLL_MAX_ATTEMPTS", 24)) # for 24 hours
|
||||
BATCH_TPD_WINDOW_SECONDS: Final = 86400
|
||||
|
|
|
|||
|
|
@ -46324,6 +46324,16 @@
|
|||
"/v1/audio/speech"
|
||||
]
|
||||
},
|
||||
"transcribe/StartTranscriptionJob": {
|
||||
"input_cost_per_second": 0.0001,
|
||||
"litellm_provider": "transcribe",
|
||||
"mode": "audio_transcription",
|
||||
"output_cost_per_second": 0.0,
|
||||
"source": "https://aws.amazon.com/transcribe/pricing/",
|
||||
"metadata": {
|
||||
"notes": "Amazon Transcribe standard batch transcription, billed per second of audio with no minimum. Same rate in every region of the AWS Price List offer file for transcribe (checked 2026-09-17)"
|
||||
}
|
||||
},
|
||||
"aws_polly/standard": {
|
||||
"input_cost_per_character": 4e-06,
|
||||
"litellm_provider": "aws_polly",
|
||||
|
|
|
|||
|
|
@ -208,6 +208,7 @@ LAZY_FEATURES: Final[tuple[LazyFeature, ...]] = (
|
|||
"/nvidia_nim/",
|
||||
"/openai/",
|
||||
"/openai_passthrough/",
|
||||
"/transcribe",
|
||||
"/typesafe/",
|
||||
"/vertex-ai/",
|
||||
"/vertex_ai/",
|
||||
|
|
|
|||
|
|
@ -20373,6 +20373,77 @@
|
|||
]
|
||||
}
|
||||
},
|
||||
"/transcribe": {
|
||||
"post": {
|
||||
"description": "AWS-SDK-shaped pass-through for Amazon Transcribe: point the SDK's `endpoint_url`\nat `/transcribe` and the operation is read from the `X-Amz-Target` header, per the\nAWS JSON 1.1 protocol.\n\n[Docs](https://docs.litellm.ai/docs/pass_through/transcribe)",
|
||||
"operationId": "transcribe_sdk_proxy_route_transcribe_post",
|
||||
"responses": {
|
||||
"200": {
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {}
|
||||
}
|
||||
},
|
||||
"description": "Successful Response"
|
||||
}
|
||||
},
|
||||
"security": [
|
||||
{
|
||||
"APIKeyHeader": []
|
||||
}
|
||||
],
|
||||
"summary": "Transcribe Sdk Proxy Route",
|
||||
"tags": [
|
||||
"llm_passthrough"
|
||||
]
|
||||
}
|
||||
},
|
||||
"/transcribe/{operation}": {
|
||||
"post": {
|
||||
"description": "Pass-through for the Amazon Transcribe API, e.g. `POST /transcribe/StartTranscriptionJob`.\n\nThe request body is forwarded to the AWS JSON 1.1 API and signed with SigV4 using the\nproxy's AWS credentials. Standard jobs are tagged with the calling key's owner so that\nonly that owner (or a proxy admin) can read or delete them, and keys other than proxy\nadmins may only read media from and write transcripts to the S3 buckets listed in\n`general_settings.transcribe_media_buckets`; account-wide operations\nsuch as ListTranscriptionJobs are limited to proxy admins. Streaming transcription\n(`transcribestreaming`) uses a separate HTTP/2 event-stream protocol and is not served\nby this route.\n\n[Docs](https://docs.litellm.ai/docs/pass_through/transcribe)",
|
||||
"operationId": "transcribe_proxy_route_transcribe__operation__post",
|
||||
"parameters": [
|
||||
{
|
||||
"in": "path",
|
||||
"name": "operation",
|
||||
"required": true,
|
||||
"schema": {
|
||||
"title": "Operation",
|
||||
"type": "string"
|
||||
}
|
||||
}
|
||||
],
|
||||
"responses": {
|
||||
"200": {
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {}
|
||||
}
|
||||
},
|
||||
"description": "Successful Response"
|
||||
},
|
||||
"422": {
|
||||
"content": {
|
||||
"application/json": {
|
||||
"schema": {
|
||||
"$ref": "#/components/schemas/HTTPValidationError"
|
||||
}
|
||||
}
|
||||
},
|
||||
"description": "Validation Error"
|
||||
}
|
||||
},
|
||||
"security": [
|
||||
{
|
||||
"APIKeyHeader": []
|
||||
}
|
||||
],
|
||||
"summary": "Transcribe Proxy Route",
|
||||
"tags": [
|
||||
"llm_passthrough"
|
||||
]
|
||||
}
|
||||
},
|
||||
"/typesafe/{endpoint}": {
|
||||
"delete": {
|
||||
"description": "[Docs](https://docs.litellm.ai/docs/pass_through/typesafe)",
|
||||
|
|
|
|||
|
|
@ -469,6 +469,7 @@ class LiteLLMRoutes(enum.Enum):
|
|||
mapped_pass_through_routes = [
|
||||
"/bedrock",
|
||||
"/comprehendmedical",
|
||||
"/transcribe",
|
||||
"/vertex-ai",
|
||||
"/vertex_ai",
|
||||
"/cohere",
|
||||
|
|
@ -2785,6 +2786,10 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
|
|||
default=None,
|
||||
description="Serve the OpenAI pass-through WebSocket route, which relays frames to OpenAI under the proxy's own provider credential without reading them. Off by default.",
|
||||
)
|
||||
transcribe_media_buckets: list[str] | None = Field(
|
||||
default=None,
|
||||
description="S3 bucket names that keys other than proxy admins may read media from and write transcripts to through the Amazon Transcribe pass-through. Unset means only proxy admins can start transcription jobs.",
|
||||
)
|
||||
user_header_name: str | None = Field(
|
||||
None,
|
||||
description="[DEPRECATED] Use 'user_header_mappings' instead. When set, the header value is treated as the end user id unless overridden by user_header_mappings.",
|
||||
|
|
|
|||
|
|
@ -93,6 +93,7 @@ _LLM_ROUTE_EXACT: Final[tuple[str, ...]] = (
|
|||
"/interactions", # Google Interactions create; /{id} reads and /cancel do not match
|
||||
"/v1beta/interactions",
|
||||
"/comprehendmedical", # AWS-SDK-shaped passthrough: the operation rides in the X-Amz-Target header
|
||||
"/transcribe",
|
||||
)
|
||||
|
||||
# Provider passthrough prefixes (e.g. /bedrock/..., /vertex-ai/...) carry real
|
||||
|
|
|
|||
|
|
@ -1235,7 +1235,13 @@ async def bedrock_proxy_route(
|
|||
COMPREHEND_MEDICAL_TARGET_PREFIX: Final = "ComprehendMedical_20181030"
|
||||
|
||||
|
||||
def _resolve_comprehend_medical_region() -> str | None:
|
||||
def _proxy_general_settings() -> Mapping[str, object]:
|
||||
from litellm.proxy.proxy_server import general_settings
|
||||
|
||||
return general_settings
|
||||
|
||||
|
||||
def _resolve_aws_passthrough_region() -> str | None:
|
||||
region_candidates: Final = (
|
||||
get_secret_str(secret_name="AWS_REGION_NAME"),
|
||||
get_secret_str(secret_name="AWS_REGION"),
|
||||
|
|
@ -1275,7 +1281,7 @@ async def comprehend_medical_proxy_route(
|
|||
),
|
||||
)
|
||||
|
||||
aws_region_name: Final = _resolve_comprehend_medical_region()
|
||||
aws_region_name: Final = _resolve_aws_passthrough_region()
|
||||
if aws_region_name is None:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
|
|
@ -1352,6 +1358,167 @@ async def comprehend_medical_sdk_proxy_route(
|
|||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/transcribe/{operation}",
|
||||
tags=["Amazon Transcribe Pass-through", "pass-through"], # mutable-ok: fastapi route tags must be a list
|
||||
)
|
||||
async def transcribe_proxy_route(
|
||||
operation: str,
|
||||
request: Request,
|
||||
fastapi_response: Response,
|
||||
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
|
||||
general_settings: Annotated[Mapping[str, object], Depends(_proxy_general_settings)],
|
||||
):
|
||||
"""
|
||||
Pass-through for the Amazon Transcribe API, e.g. `POST /transcribe/StartTranscriptionJob`.
|
||||
|
||||
The request body is forwarded to the AWS JSON 1.1 API and signed with SigV4 using the
|
||||
proxy's AWS credentials. Standard jobs are tagged with the calling key's owner so that
|
||||
only that owner (or a proxy admin) can read or delete them, and keys other than proxy
|
||||
admins may only read media from and write transcripts to the S3 buckets listed in
|
||||
`general_settings.transcribe_media_buckets`; account-wide operations
|
||||
such as ListTranscriptionJobs are limited to proxy admins. Streaming transcription
|
||||
(`transcribestreaming`) uses a separate HTTP/2 event-stream protocol and is not served
|
||||
by this route.
|
||||
|
||||
[Docs](https://docs.litellm.ai/docs/pass_through/transcribe)
|
||||
"""
|
||||
from .llm_provider_handlers.transcribe_passthrough_logging_handler import (
|
||||
TRANSCRIBE_CUSTOM_LLM_PROVIDER,
|
||||
TRANSCRIBE_OWNED_JOB_OPERATIONS,
|
||||
TRANSCRIBE_PRICED_OPERATION,
|
||||
TRANSCRIBE_TARGET_PREFIX,
|
||||
TranscribeRefusal,
|
||||
transcribe_admin_only_refusal,
|
||||
transcribe_cost_per_second,
|
||||
transcribe_job_access_refusal,
|
||||
transcribe_job_lookup,
|
||||
transcribe_media_buckets,
|
||||
transcribe_owned_start_request,
|
||||
transcribe_storage_refusal,
|
||||
transcribe_supported_operations,
|
||||
transcribe_unpriceable_request_reason,
|
||||
)
|
||||
|
||||
if operation not in transcribe_supported_operations():
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=(
|
||||
f"Unsupported Amazon Transcribe operation: {operation}. "
|
||||
f"Supported operations: {', '.join(sorted(transcribe_supported_operations()))}"
|
||||
),
|
||||
)
|
||||
|
||||
aws_region_name: Final = _resolve_aws_passthrough_region()
|
||||
if aws_region_name is None:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="AWS region not found. Set AWS_REGION_NAME in the proxy environment.",
|
||||
)
|
||||
|
||||
try:
|
||||
data: Final = await _json_request_body(request)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=f"Request body must be valid JSON: {e}")
|
||||
|
||||
if not isinstance(data, dict):
|
||||
raise HTTPException(status_code=400, detail="Request body must be a JSON object")
|
||||
if "stream" in data:
|
||||
raise HTTPException(status_code=400, detail="'stream' is not an Amazon Transcribe request member")
|
||||
unpriceable_reason: Final = transcribe_unpriceable_request_reason(operation, data, transcribe_cost_per_second())
|
||||
if unpriceable_reason is not None:
|
||||
raise HTTPException(status_code=400, detail=unpriceable_reason)
|
||||
admin_only_refusal: Final = transcribe_admin_only_refusal(operation, user_api_key_dict)
|
||||
if admin_only_refusal is not None:
|
||||
raise HTTPException(status_code=admin_only_refusal.status_code, detail=admin_only_refusal.detail)
|
||||
storage_refusal: Final = (
|
||||
transcribe_storage_refusal(data, transcribe_media_buckets(general_settings), user_api_key_dict)
|
||||
if operation == TRANSCRIBE_PRICED_OPERATION
|
||||
else None
|
||||
)
|
||||
if storage_refusal is not None:
|
||||
raise HTTPException(status_code=storage_refusal.status_code, detail=storage_refusal.detail)
|
||||
request_body: Final = (
|
||||
transcribe_owned_start_request(data, user_api_key_dict) if operation == TRANSCRIBE_PRICED_OPERATION else data
|
||||
)
|
||||
if isinstance(request_body, TranscribeRefusal):
|
||||
raise HTTPException(status_code=request_body.status_code, detail=request_body.detail)
|
||||
access_refusal: Final = (
|
||||
await transcribe_job_access_refusal(
|
||||
data.get("TranscriptionJobName"), user_api_key_dict, transcribe_job_lookup(aws_region_name)
|
||||
)
|
||||
if operation in TRANSCRIBE_OWNED_JOB_OPERATIONS
|
||||
else None
|
||||
)
|
||||
if access_refusal is not None:
|
||||
raise HTTPException(status_code=access_refusal.status_code, detail=access_refusal.detail)
|
||||
|
||||
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM, run_aws_signing, sign_aws_json_post
|
||||
|
||||
target_url: Final = f"https://transcribe.{aws_region_name}.{get_aws_dns_suffix(aws_region_name)}/"
|
||||
prepped: Final = await run_aws_signing(
|
||||
sign_aws_json_post,
|
||||
get_credentials=partial(BaseAWSLLM().get_credentials, aws_region_name=aws_region_name),
|
||||
service_name="transcribe",
|
||||
aws_region_name=aws_region_name,
|
||||
url=target_url,
|
||||
body=json.dumps(request_body),
|
||||
headers=MappingProxyType(
|
||||
{
|
||||
"Content-Type": "application/x-amz-json-1.1",
|
||||
"X-Amz-Target": f"{TRANSCRIBE_TARGET_PREFIX}.{operation}",
|
||||
}
|
||||
),
|
||||
)
|
||||
|
||||
endpoint_func: Final = create_pass_through_route(
|
||||
endpoint=operation,
|
||||
target=str(prepped.url),
|
||||
custom_headers=prepped.headers,
|
||||
custom_llm_provider=TRANSCRIBE_CUSTOM_LLM_PROVIDER,
|
||||
)
|
||||
setattr(request.state, LITELLM_PASS_THROUGH_CUSTOM_BODY_STATE_KEY, request_body)
|
||||
setattr(request.state, LITELLM_PASS_THROUGH_RAW_BODY_STATE_KEY, prepped.body)
|
||||
return await endpoint_func(request, fastapi_response, user_api_key_dict)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/transcribe",
|
||||
tags=["Amazon Transcribe Pass-through", "pass-through"], # mutable-ok: fastapi route tags must be a list
|
||||
)
|
||||
async def transcribe_sdk_proxy_route(
|
||||
request: Request,
|
||||
fastapi_response: Response,
|
||||
user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)],
|
||||
general_settings: Annotated[Mapping[str, object], Depends(_proxy_general_settings)],
|
||||
):
|
||||
"""
|
||||
AWS-SDK-shaped pass-through for Amazon Transcribe: point the SDK's `endpoint_url`
|
||||
at `/transcribe` and the operation is read from the `X-Amz-Target` header, per the
|
||||
AWS JSON 1.1 protocol.
|
||||
|
||||
[Docs](https://docs.litellm.ai/docs/pass_through/transcribe)
|
||||
"""
|
||||
from .llm_provider_handlers.transcribe_passthrough_logging_handler import (
|
||||
TRANSCRIBE_TARGET_PREFIX,
|
||||
)
|
||||
|
||||
target_header: Final = request.headers.get("x-amz-target", "")
|
||||
target_prefix, _, operation = target_header.partition(".")
|
||||
if target_prefix != TRANSCRIBE_TARGET_PREFIX or not operation:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"Expected an X-Amz-Target header of the form {TRANSCRIBE_TARGET_PREFIX}.<Operation>",
|
||||
)
|
||||
return await transcribe_proxy_route(
|
||||
operation=operation,
|
||||
request=request,
|
||||
fastapi_response=fastapi_response,
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
general_settings=general_settings,
|
||||
)
|
||||
|
||||
|
||||
def _resolve_vertex_model_from_router(
|
||||
model_id: str,
|
||||
llm_router: litellm.Router | None,
|
||||
|
|
@ -2623,12 +2790,6 @@ class _OpenAIWebsocketRelay(Protocol):
|
|||
) -> None: ...
|
||||
|
||||
|
||||
def _proxy_general_settings() -> Mapping[str, object]:
|
||||
from litellm.proxy.proxy_server import general_settings
|
||||
|
||||
return general_settings
|
||||
|
||||
|
||||
def _openai_websocket_relay() -> _OpenAIWebsocketRelay:
|
||||
return websocket_passthrough_request
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,733 @@
|
|||
import asyncio
|
||||
import json
|
||||
import math
|
||||
import tempfile
|
||||
from collections.abc import Awaitable, Callable, Mapping
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from email.utils import parsedate_to_datetime
|
||||
from functools import lru_cache, partial
|
||||
from pathlib import Path
|
||||
from types import MappingProxyType
|
||||
from typing import IO, Final, Protocol, TypeAlias
|
||||
from urllib.parse import quote
|
||||
|
||||
import httpx
|
||||
import soundfile
|
||||
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
|
||||
from typing_extensions import ReadOnly, TypedDict
|
||||
|
||||
import litellm
|
||||
from litellm._logging import verbose_proxy_logger
|
||||
from litellm.constants import (
|
||||
TRANSCRIBE_JOB_MAX_POLLING_ATTEMPTS,
|
||||
TRANSCRIBE_JOB_POLLING_INTERVAL_SECONDS,
|
||||
TRANSCRIBE_MAX_MEDIA_BYTES,
|
||||
TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS,
|
||||
TRANSCRIBE_MEASURABLE_MEDIA_FORMATS,
|
||||
TRANSCRIBE_MEDIA_DOWNLOAD_CONCURRENCY,
|
||||
TRANSCRIBE_MEDIA_FETCH_ATTEMPTS,
|
||||
TRANSCRIBE_MEDIA_LAST_MODIFIED_TOLERANCE_SECONDS,
|
||||
)
|
||||
from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
from litellm.litellm_core_utils.litellm_logging import (
|
||||
get_standard_logging_object_payload,
|
||||
)
|
||||
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
|
||||
from litellm.proxy._types import (
|
||||
PassThroughEndpointLoggingResultValues,
|
||||
PassThroughEndpointLoggingTypedDict,
|
||||
UserAPIKeyAuth,
|
||||
)
|
||||
from litellm.proxy.common_utils.resource_ownership import (
|
||||
get_primary_resource_owner_scope,
|
||||
is_proxy_admin,
|
||||
user_can_access_resource_owner,
|
||||
)
|
||||
from litellm.types.llms.custom_http import httpxSpecialProvider
|
||||
from litellm.types.utils import StandardPassThroughResponseObject
|
||||
|
||||
TRANSCRIBE_TARGET_PREFIX: Final = "Transcribe"
|
||||
TRANSCRIBE_CUSTOM_LLM_PROVIDER: Final = "transcribe"
|
||||
TRANSCRIBE_PRICED_OPERATION: Final = "StartTranscriptionJob"
|
||||
TRANSCRIBE_PRICED_MODEL: Final = f"{TRANSCRIBE_CUSTOM_LLM_PROVIDER}/{TRANSCRIBE_PRICED_OPERATION}"
|
||||
TRANSCRIBE_UNPRICED_OPERATIONS: Final = frozenset(
|
||||
{"StartCallAnalyticsJob", "StartMedicalScribeJob", "StartMedicalTranscriptionJob"}
|
||||
)
|
||||
TRANSCRIBE_SURCHARGE_MEMBERS: Final = ("ContentRedaction", "ToxicityDetection")
|
||||
TRANSCRIBE_TERMINAL_JOB_STATUSES: Final = frozenset({"COMPLETED", "FAILED"})
|
||||
TRANSCRIBE_MISSING_JOB_ERRORS: Final = frozenset({"BadRequestException", "NotFoundException"})
|
||||
TRANSCRIBE_OWNER_TAG: Final = "litellm-owner"
|
||||
TRANSCRIBE_OWNED_JOB_OPERATIONS: Final = frozenset({"GetTranscriptionJob", "DeleteTranscriptionJob"})
|
||||
TRANSCRIBE_MEDIA_BUCKETS_SETTING: Final = "transcribe_media_buckets"
|
||||
TRANSCRIBE_ROLE_MEMBERS: Final = ("DataAccessRoleArn", "JobExecutionSettings")
|
||||
TRANSCRIBE_MEDIA_URI_MEMBERS: Final = ("MediaFileUri", "RedactedMediaFileUri")
|
||||
|
||||
JobLookup: TypeAlias = Callable[[str], Awaitable[Mapping[str, object]]] # mutable-ok: Callable parameter syntax
|
||||
MediaDurationProbe: TypeAlias = Callable[[str, float], Awaitable[float | None]] # mutable-ok: Callable parameter syntax
|
||||
|
||||
|
||||
class GetTranscriptionJobRequest(TypedDict):
|
||||
TranscriptionJobName: ReadOnly[str]
|
||||
|
||||
|
||||
class _MediaRef(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
MediaFileUri: str | None = None
|
||||
|
||||
|
||||
class _JobTag(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
Key: str | None = None
|
||||
Value: str | None = None
|
||||
|
||||
|
||||
class TranscriptionJobRecord(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
TranscriptionJobStatus: str | None = None
|
||||
CreationTime: float | None = None
|
||||
Media: _MediaRef | None = None
|
||||
Tags: tuple[_JobTag, ...] = ()
|
||||
|
||||
|
||||
class _TranscriptionJobResponse(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
TranscriptionJob: TranscriptionJobRecord | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class MissingJob:
|
||||
"""Transcribe no longer knows the job, so polling it again can never reach a terminal status."""
|
||||
|
||||
|
||||
StartedJob: TypeAlias = TranscriptionJobRecord | None
|
||||
JobPricer: TypeAlias = Callable[[str, str, float, StartedJob], Awaitable[float]] # mutable-ok: Callable params
|
||||
|
||||
|
||||
class _PricedCostMapEntry(BaseModel):
|
||||
model_config = ConfigDict(frozen=True, strict=True)
|
||||
input_cost_per_second: float
|
||||
|
||||
|
||||
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object])
|
||||
_JSON_OBJECTS: Final = TypeAdapter(tuple[Mapping[str, object], ...])
|
||||
_BUCKET_NAMES: Final = TypeAdapter(frozenset[str])
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class TranscribeRefusal:
|
||||
status_code: int
|
||||
detail: str
|
||||
|
||||
|
||||
class PassThroughLogDispatch(Protocol):
|
||||
def __call__(
|
||||
self,
|
||||
*,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
standard_logging_response_object: PassThroughEndpointLoggingResultValues | None,
|
||||
result: str,
|
||||
start_time: datetime,
|
||||
end_time: datetime,
|
||||
cache_hit: bool,
|
||||
**kwargs: object, # kwargs-ok: mirrors the shared pass-through logging dispatch signature
|
||||
) -> Awaitable[None]: ...
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def transcribe_supported_operations() -> frozenset[str]:
|
||||
"""
|
||||
Operation names of the Amazon Transcribe JSON 1.1 API, read from the botocore
|
||||
service model so the allowlist tracks the installed SDK instead of a hand-typed copy.
|
||||
"""
|
||||
from botocore.session import get_session
|
||||
|
||||
return frozenset(get_session().get_service_model("transcribe").operation_names)
|
||||
|
||||
|
||||
def transcribe_cost_per_second() -> float | None:
|
||||
try:
|
||||
return _PricedCostMapEntry.model_validate(litellm.model_cost.get(TRANSCRIBE_PRICED_MODEL)).input_cost_per_second
|
||||
except ValidationError:
|
||||
return None
|
||||
|
||||
|
||||
def transcribe_unpriceable_request_reason(
|
||||
operation: str,
|
||||
request_body: Mapping[str, object],
|
||||
cost_per_second: float | None,
|
||||
) -> str | None:
|
||||
if operation in TRANSCRIBE_UNPRICED_OPERATIONS:
|
||||
return (
|
||||
f"{operation} is billed per second of audio at a rate LiteLLM does not price yet, so it cannot be"
|
||||
f" submitted through this route; only {TRANSCRIBE_PRICED_OPERATION} is priced and budgeted"
|
||||
)
|
||||
if operation != TRANSCRIBE_PRICED_OPERATION:
|
||||
return None
|
||||
if cost_per_second is None:
|
||||
return (
|
||||
f"{TRANSCRIBE_PRICED_MODEL} has no input_cost_per_second in the LiteLLM model cost map, so billable"
|
||||
" transcription jobs cannot be submitted through this route"
|
||||
)
|
||||
surcharges: Final = tuple(m for m in TRANSCRIBE_SURCHARGE_MEMBERS if m in request_body) + tuple(
|
||||
_custom_language_model_members(request_body)
|
||||
)
|
||||
if surcharges:
|
||||
return (
|
||||
f"{TRANSCRIBE_PRICED_OPERATION} with {', '.join(surcharges)} adds a per-second surcharge LiteLLM does not"
|
||||
" price yet; remove it to submit the job through this route"
|
||||
)
|
||||
if requested_media_format(request_body) not in TRANSCRIBE_MEASURABLE_MEDIA_FORMATS:
|
||||
return (
|
||||
"LiteLLM bills a transcription job by reading the length of the media file, which it can only do for"
|
||||
f" {', '.join(sorted(TRANSCRIBE_MEASURABLE_MEDIA_FORMATS))}; set MediaFormat to one of those or point"
|
||||
" Media.MediaFileUri at a file with that extension"
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def _custom_language_model_members(request_body: Mapping[str, object]) -> tuple[str, ...]:
|
||||
model_settings: Final = request_body.get("ModelSettings")
|
||||
language_id_settings: Final = request_body.get("LanguageIdSettings")
|
||||
from_model_settings: Final = (
|
||||
("ModelSettings.LanguageModelName",)
|
||||
if isinstance(model_settings, Mapping) and "LanguageModelName" in model_settings
|
||||
else ()
|
||||
)
|
||||
from_language_id: Final = (
|
||||
tuple(
|
||||
f"LanguageIdSettings.{language}.LanguageModelName"
|
||||
for language, settings in _JSON_OBJECT.validate_python(language_id_settings).items()
|
||||
if isinstance(settings, Mapping) and "LanguageModelName" in settings
|
||||
)
|
||||
if isinstance(language_id_settings, Mapping)
|
||||
else ()
|
||||
)
|
||||
return from_model_settings + from_language_id
|
||||
|
||||
|
||||
def requested_media_format(request_body: Mapping[str, object]) -> str | None:
|
||||
media_format: Final = request_body.get("MediaFormat")
|
||||
if isinstance(media_format, str):
|
||||
return media_format.lower()
|
||||
media: Final = request_body.get("Media")
|
||||
media_uri: Final = _JSON_OBJECT.validate_python(media).get("MediaFileUri") if isinstance(media, Mapping) else None
|
||||
if not isinstance(media_uri, str):
|
||||
return None
|
||||
path: Final = httpx.URL(media_uri).path if "://" in media_uri else media_uri
|
||||
_, dot, suffix = path.rpartition(".")
|
||||
return suffix.lower() if dot else None
|
||||
|
||||
|
||||
def transcribe_admin_only_refusal(operation: str, user_api_key_dict: UserAPIKeyAuth) -> TranscribeRefusal | None:
|
||||
if (
|
||||
operation == TRANSCRIBE_PRICED_OPERATION
|
||||
or operation in TRANSCRIBE_OWNED_JOB_OPERATIONS
|
||||
or is_proxy_admin(user_api_key_dict)
|
||||
):
|
||||
return None
|
||||
return TranscribeRefusal(
|
||||
403,
|
||||
f"{operation} reaches every Amazon Transcribe resource in the AWS account, so only a proxy admin may call it;"
|
||||
f" other keys may {TRANSCRIBE_PRICED_OPERATION} and {' or '.join(sorted(TRANSCRIBE_OWNED_JOB_OPERATIONS))}"
|
||||
" for the jobs they started",
|
||||
)
|
||||
|
||||
|
||||
def transcribe_media_buckets(general_settings: Mapping[str, object]) -> frozenset[str] | None:
|
||||
try:
|
||||
return _BUCKET_NAMES.validate_python(general_settings.get(TRANSCRIBE_MEDIA_BUCKETS_SETTING))
|
||||
except ValidationError:
|
||||
return None
|
||||
|
||||
|
||||
def s3_bucket_name(uri: object) -> str | None:
|
||||
if not isinstance(uri, str) or not uri.startswith("s3://"):
|
||||
return None
|
||||
bucket, _, _ = uri.removeprefix("s3://").partition("/")
|
||||
return bucket or None
|
||||
|
||||
|
||||
def transcribe_storage_refusal(
|
||||
request_body: Mapping[str, object],
|
||||
allowed_buckets: frozenset[str] | None,
|
||||
user_api_key_dict: UserAPIKeyAuth,
|
||||
) -> TranscribeRefusal | None:
|
||||
"""
|
||||
Transcribe reads the media and writes the transcript with the proxy's own AWS credentials, so a
|
||||
non-admin key may only point a job at buckets the operator listed; otherwise any object those
|
||||
credentials can reach could be transcribed and read back through the caller's own job.
|
||||
"""
|
||||
if is_proxy_admin(user_api_key_dict):
|
||||
return None
|
||||
if allowed_buckets is None:
|
||||
return TranscribeRefusal(
|
||||
403,
|
||||
f"general_settings.{TRANSCRIBE_MEDIA_BUCKETS_SETTING} is not a list of S3 bucket names, so only a proxy"
|
||||
f" admin may {TRANSCRIBE_PRICED_OPERATION}; list the buckets other keys may read media from and write"
|
||||
" transcripts to",
|
||||
)
|
||||
roles: Final = tuple(m for m in TRANSCRIBE_ROLE_MEMBERS if m in request_body)
|
||||
if roles:
|
||||
return TranscribeRefusal(
|
||||
403,
|
||||
f"{', '.join(roles)} would run the job under a role other than the proxy's own AWS credentials, so"
|
||||
" only a proxy admin may set it",
|
||||
)
|
||||
media: Final = request_body.get("Media")
|
||||
media_uris: Final = (
|
||||
tuple((f"Media.{m}", s3_bucket_name(media.get(m))) for m in TRANSCRIBE_MEDIA_URI_MEMBERS if m in media)
|
||||
if isinstance(media, Mapping)
|
||||
else ()
|
||||
)
|
||||
output: Final = request_body.get("OutputBucketName")
|
||||
locations: Final = media_uris + (
|
||||
(("OutputBucketName", output if isinstance(output, str) else None),)
|
||||
if "OutputBucketName" in request_body
|
||||
else ()
|
||||
)
|
||||
offending: Final = tuple(member for member, bucket in locations if bucket not in allowed_buckets)
|
||||
if offending:
|
||||
return TranscribeRefusal(
|
||||
403,
|
||||
f"{', '.join(offending)} must name one of the S3 buckets in general_settings."
|
||||
f"{TRANSCRIBE_MEDIA_BUCKETS_SETTING} ({', '.join(sorted(allowed_buckets))}), as s3://bucket/key for media",
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def transcribe_owned_start_request(
|
||||
request_body: Mapping[str, object], user_api_key_dict: UserAPIKeyAuth
|
||||
) -> dict[str, object] | TranscribeRefusal:
|
||||
owner: Final = get_primary_resource_owner_scope(user_api_key_dict)
|
||||
if owner is None:
|
||||
return TranscribeRefusal(400, "The calling key has no identity to record as the owner of the transcription job")
|
||||
try:
|
||||
tags: Final = _JSON_OBJECTS.validate_python(request_body.get("Tags", ()))
|
||||
except ValidationError:
|
||||
return TranscribeRefusal(400, "Tags must be a list of objects with Key and Value members")
|
||||
if any(tag.get("Key") == TRANSCRIBE_OWNER_TAG for tag in tags):
|
||||
return TranscribeRefusal(
|
||||
400, f"The {TRANSCRIBE_OWNER_TAG} tag is assigned by LiteLLM and cannot be supplied by the caller"
|
||||
)
|
||||
owner_tag: Final = _JobTag(Key=TRANSCRIBE_OWNER_TAG, Value=owner).model_dump()
|
||||
return {**request_body, "Tags": (*tags, owner_tag)} # mutable-ok: json.dumps and the body state key take a dict
|
||||
|
||||
|
||||
async def transcribe_job_access_refusal(
|
||||
job_name: object, user_api_key_dict: UserAPIKeyAuth, get_job: JobLookup
|
||||
) -> TranscribeRefusal | None:
|
||||
if is_proxy_admin(user_api_key_dict):
|
||||
return None
|
||||
if not isinstance(job_name, str):
|
||||
return TranscribeRefusal(400, "TranscriptionJobName must be a string")
|
||||
not_found: Final = TranscribeRefusal(
|
||||
404, f"No transcription job named {job_name} was started through this proxy by the calling key"
|
||||
)
|
||||
try:
|
||||
job: Final = _TranscriptionJobResponse.model_validate(await get_job(job_name)).TranscriptionJob
|
||||
except Exception as e: # noqa: BLE001 # a job that cannot be read cannot be shown to belong to the caller
|
||||
verbose_proxy_logger.warning("Looking up Transcribe job %s for an ownership check failed: %s", job_name, e)
|
||||
return not_found
|
||||
owner: Final = (
|
||||
next((tag.Value for tag in job.Tags if tag.Key == TRANSCRIBE_OWNER_TAG), None) if job is not None else None
|
||||
)
|
||||
return None if user_can_access_resource_owner(owner, user_api_key_dict) else not_found
|
||||
|
||||
|
||||
def transcription_job_cost(audio_seconds: float, cost_per_second: float) -> float:
|
||||
return math.ceil(audio_seconds) * cost_per_second
|
||||
|
||||
|
||||
def transcribe_max_job_cost(cost_per_second: float) -> float:
|
||||
return transcription_job_cost(TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS, cost_per_second)
|
||||
|
||||
|
||||
def started_transcription_job(response_body: Mapping[str, object] | None) -> TranscriptionJobRecord | None:
|
||||
try:
|
||||
return _TranscriptionJobResponse.model_validate(response_body).TranscriptionJob
|
||||
except ValidationError:
|
||||
return None
|
||||
|
||||
|
||||
def aws_error_type(response: httpx.Response) -> str | None:
|
||||
try:
|
||||
error_type: Final = _JSON_OBJECT.validate_python(response.json()).get("__type")
|
||||
except (ValueError, ValidationError):
|
||||
return None
|
||||
return error_type.rsplit("#", 1)[-1] if isinstance(error_type, str) else None
|
||||
|
||||
|
||||
async def _poll_transcription_job(job_name: str, get_job: JobLookup) -> TranscriptionJobRecord | MissingJob | None:
|
||||
try:
|
||||
job: Final = _TranscriptionJobResponse.model_validate(await get_job(job_name)).TranscriptionJob
|
||||
except httpx.HTTPStatusError as e:
|
||||
if aws_error_type(e.response) in TRANSCRIBE_MISSING_JOB_ERRORS:
|
||||
verbose_proxy_logger.warning(
|
||||
"Transcribe job %s no longer exists, pricing the media it was started with", job_name
|
||||
)
|
||||
return MissingJob()
|
||||
verbose_proxy_logger.warning("Polling Transcribe job %s failed, retrying: %s", job_name, e)
|
||||
return None
|
||||
except Exception as e: # noqa: BLE001 # a failed poll is retried on the next tick instead of ending pricing
|
||||
verbose_proxy_logger.warning("Polling Transcribe job %s failed, retrying: %s", job_name, e)
|
||||
return None
|
||||
return job if job is not None and job.TranscriptionJobStatus in TRANSCRIBE_TERMINAL_JOB_STATUSES else None
|
||||
|
||||
|
||||
async def await_transcription_job(
|
||||
job_name: str,
|
||||
get_job: JobLookup,
|
||||
sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
|
||||
max_attempts: int = TRANSCRIBE_JOB_MAX_POLLING_ATTEMPTS,
|
||||
) -> TranscriptionJobRecord | MissingJob | None:
|
||||
for _ in range(max_attempts):
|
||||
job = await _poll_transcription_job(job_name, get_job)
|
||||
if job is not None:
|
||||
return job
|
||||
await sleep(TRANSCRIBE_JOB_POLLING_INTERVAL_SECONDS)
|
||||
return None
|
||||
|
||||
|
||||
async def measure_media_seconds(
|
||||
media_uri: str,
|
||||
job_created_at: float,
|
||||
media_seconds: MediaDurationProbe,
|
||||
sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
|
||||
attempts: int = TRANSCRIBE_MEDIA_FETCH_ATTEMPTS,
|
||||
) -> float | None:
|
||||
for attempt in range(1, attempts + 1):
|
||||
try:
|
||||
return await media_seconds(media_uri, job_created_at)
|
||||
except Exception as e: # noqa: BLE001 # the media is retried, then charged at the maximum if still unreadable
|
||||
verbose_proxy_logger.warning("Measuring Transcribe media %s failed (attempt %d): %s", media_uri, attempt, e)
|
||||
if attempt < attempts:
|
||||
await sleep(TRANSCRIBE_JOB_POLLING_INTERVAL_SECONDS)
|
||||
return None
|
||||
|
||||
|
||||
async def price_transcription_job(
|
||||
job_name: str,
|
||||
cost_per_second: float,
|
||||
get_job: JobLookup,
|
||||
media_seconds: MediaDurationProbe,
|
||||
sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
|
||||
max_attempts: int = TRANSCRIBE_JOB_MAX_POLLING_ATTEMPTS,
|
||||
started_job: TranscriptionJobRecord | None = None,
|
||||
) -> float:
|
||||
"""
|
||||
Amazon Transcribe bills every second of the media file, silence included, and reports no
|
||||
duration itself, so the job is polled to completion and the media it transcribed is measured.
|
||||
The measurement only counts when the object has not been rewritten since the job was created,
|
||||
which is what ties it to the bytes Transcribe read. A job deleted before it is polled is
|
||||
measured from the media named in its StartTranscriptionJob response. Anything that stops the
|
||||
duration from being read is charged as the longest media AWS accepts.
|
||||
"""
|
||||
outcome: Final = await await_transcription_job(job_name, get_job, sleep=sleep, max_attempts=max_attempts)
|
||||
if outcome is None:
|
||||
verbose_proxy_logger.warning("Transcribe job %s did not finish while polling, charging maximum", job_name)
|
||||
return transcribe_max_job_cost(cost_per_second)
|
||||
if isinstance(outcome, TranscriptionJobRecord) and outcome.TranscriptionJobStatus == "FAILED":
|
||||
return 0.0
|
||||
job: Final = outcome if isinstance(outcome, TranscriptionJobRecord) else started_job
|
||||
media_uri: Final = job.Media.MediaFileUri if job is not None and job.Media is not None else None
|
||||
if job is None or media_uri is None or job.CreationTime is None:
|
||||
return transcribe_max_job_cost(cost_per_second)
|
||||
audio_seconds: Final = await measure_media_seconds(media_uri, job.CreationTime, media_seconds, sleep=sleep)
|
||||
if audio_seconds is None:
|
||||
return transcribe_max_job_cost(cost_per_second)
|
||||
return transcription_job_cost(audio_seconds, cost_per_second)
|
||||
|
||||
|
||||
def _as_json_object(response: httpx.Response) -> Mapping[str, object]:
|
||||
return _JSON_OBJECT.validate_python(response.raise_for_status().json())
|
||||
|
||||
|
||||
def transcribe_job_lookup(aws_region_name: str) -> JobLookup:
|
||||
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM, run_aws_signing, sign_aws_json_post
|
||||
|
||||
url: Final = f"https://transcribe.{aws_region_name}.{get_aws_dns_suffix(aws_region_name)}/"
|
||||
headers: Final = MappingProxyType(
|
||||
{
|
||||
"Content-Type": "application/x-amz-json-1.1",
|
||||
"X-Amz-Target": f"{TRANSCRIBE_TARGET_PREFIX}.GetTranscriptionJob",
|
||||
}
|
||||
)
|
||||
|
||||
async def get_job(job_name: str) -> Mapping[str, object]:
|
||||
body: Final[GetTranscriptionJobRequest] = {"TranscriptionJobName": job_name}
|
||||
payload: Final = json.dumps(body)
|
||||
prepped: Final = await run_aws_signing(
|
||||
sign_aws_json_post,
|
||||
get_credentials=partial(BaseAWSLLM().get_credentials, aws_region_name=aws_region_name),
|
||||
service_name="transcribe",
|
||||
aws_region_name=aws_region_name,
|
||||
url=url,
|
||||
body=payload,
|
||||
headers=headers,
|
||||
)
|
||||
client: Final = get_async_httpx_client(llm_provider=httpxSpecialProvider.PassThroughEndpoint)
|
||||
signed_headers: Final = dict(prepped.headers.items()) # mutable-ok: AsyncHTTPHandler.post takes a dict
|
||||
return _as_json_object(await client.post(str(prepped.url), data=payload, headers=signed_headers))
|
||||
|
||||
return get_job
|
||||
|
||||
|
||||
def s3_media_url(media_uri: str, aws_region_name: str) -> str | None:
|
||||
"""
|
||||
Transcribe accepts media as s3://bucket/key or as an https S3 URL; the bucket is required to
|
||||
live in the job's region, so the s3 form maps onto that region's endpoint. Buckets with dots in
|
||||
their name use the path-style form because they cannot match the virtual-hosted wildcard
|
||||
certificate. The proxy's AWS signature is only ever sent to that partition's own hosts.
|
||||
"""
|
||||
dns_suffix: Final = get_aws_dns_suffix(aws_region_name)
|
||||
if not media_uri.startswith("s3://"):
|
||||
url: Final = httpx.URL(media_uri)
|
||||
return media_uri if url.scheme == "https" and url.host.endswith(f".{dns_suffix}") else None
|
||||
bucket, _, key = media_uri.removeprefix("s3://").partition("/")
|
||||
if "." in bucket:
|
||||
return f"https://s3.{aws_region_name}.{dns_suffix}/{bucket}/{quote(key)}"
|
||||
return f"https://{bucket}.s3.{aws_region_name}.{dns_suffix}/{quote(key)}"
|
||||
|
||||
|
||||
def media_predates_job(headers: Mapping[str, str], job_created_at: float) -> bool:
|
||||
try:
|
||||
modified_at: Final = parsedate_to_datetime(headers["last-modified"]).timestamp()
|
||||
except (KeyError, TypeError, ValueError):
|
||||
return False
|
||||
return modified_at <= job_created_at + TRANSCRIBE_MEDIA_LAST_MODIFIED_TOLERANCE_SECONDS
|
||||
|
||||
|
||||
async def write_media_within_limit(response: httpx.Response, media_file: IO[bytes], max_bytes: int) -> bool:
|
||||
if int(response.headers.get("content-length", "0")) > max_bytes:
|
||||
return False
|
||||
async for chunk in response.aiter_bytes():
|
||||
_ = media_file.write(chunk)
|
||||
if media_file.tell() > max_bytes:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def media_file_seconds(path: Path) -> float | None:
|
||||
try:
|
||||
with soundfile.SoundFile(str(path)) as audio:
|
||||
return len(audio) / audio.samplerate
|
||||
except (RuntimeError, ValueError, OSError) as e:
|
||||
verbose_proxy_logger.warning("Transcribe media could not be decoded for its duration: %s", e)
|
||||
return None
|
||||
|
||||
|
||||
def transcribe_media_duration_probe(aws_region_name: str, download_slots: asyncio.Semaphore) -> MediaDurationProbe:
|
||||
from botocore.auth import S3SigV4Auth
|
||||
from botocore.awsrequest import AWSRequest
|
||||
|
||||
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM, run_aws_signing
|
||||
|
||||
def sign_s3_get(url: str) -> dict[str, str]: # mutable-ok: httpx request headers take a dict
|
||||
aws_request: Final = AWSRequest(method="GET", url=url)
|
||||
credentials: Final = BaseAWSLLM().get_credentials(aws_region_name=aws_region_name)
|
||||
S3SigV4Auth(credentials, "s3", aws_region_name).add_auth(aws_request)
|
||||
return dict(aws_request.prepare().headers.items()) # mutable-ok: httpx request headers take a dict
|
||||
|
||||
async def media_seconds(media_uri: str, job_created_at: float) -> float | None:
|
||||
url: Final = s3_media_url(media_uri, aws_region_name)
|
||||
if url is None:
|
||||
return None
|
||||
headers: Final = await run_aws_signing(sign_s3_get, url)
|
||||
client: Final = get_async_httpx_client(llm_provider=httpxSpecialProvider.PassThroughEndpoint).client
|
||||
async with download_slots:
|
||||
with tempfile.NamedTemporaryFile() as media_file:
|
||||
async with client.stream("GET", url, headers=headers) as response:
|
||||
_ = response.raise_for_status()
|
||||
if not media_predates_job(response.headers, job_created_at):
|
||||
verbose_proxy_logger.warning(
|
||||
"Transcribe media %s was rewritten after the job was created, charging maximum", media_uri
|
||||
)
|
||||
return None
|
||||
if not await write_media_within_limit(response, media_file, TRANSCRIBE_MAX_MEDIA_BYTES):
|
||||
verbose_proxy_logger.warning(
|
||||
"Transcribe media %s exceeds the size cap, charging maximum", media_uri
|
||||
)
|
||||
return None
|
||||
media_file.flush()
|
||||
return await asyncio.to_thread(media_file_seconds, Path(media_file.name))
|
||||
|
||||
return media_seconds
|
||||
|
||||
|
||||
async def price_transcription_job_live(
|
||||
job_name: str,
|
||||
aws_region_name: str,
|
||||
cost_per_second: float,
|
||||
started_job: TranscriptionJobRecord | None,
|
||||
download_slots: asyncio.Semaphore,
|
||||
) -> float:
|
||||
try:
|
||||
return await price_transcription_job(
|
||||
job_name,
|
||||
cost_per_second,
|
||||
get_job=transcribe_job_lookup(aws_region_name),
|
||||
media_seconds=transcribe_media_duration_probe(aws_region_name, download_slots),
|
||||
started_job=started_job,
|
||||
)
|
||||
except Exception as e: # noqa: BLE001 # an unreadable job must still be charged, so fail closed at the maximum
|
||||
verbose_proxy_logger.exception("Pricing Transcribe job %s failed, charging maximum: %s", job_name, e)
|
||||
return transcribe_max_job_cost(cost_per_second)
|
||||
|
||||
|
||||
class TranscribePassthroughLoggingHandler:
|
||||
def __init__(self, job_pricer: JobPricer | None = None) -> None:
|
||||
self._job_pricer: Final = (
|
||||
job_pricer
|
||||
if job_pricer is not None
|
||||
else partial(
|
||||
price_transcription_job_live,
|
||||
download_slots=asyncio.Semaphore(TRANSCRIBE_MEDIA_DOWNLOAD_CONCURRENCY),
|
||||
)
|
||||
)
|
||||
self._pricing_tasks: Final[set[asyncio.Task[None]]] = set() # mutable-ok: asyncio holds tasks weakly
|
||||
|
||||
@staticmethod
|
||||
def _operation_from_response(httpx_response: httpx.Response) -> str:
|
||||
headers: Final[Mapping[str, str]] = httpx_response.request.headers
|
||||
target: Final = headers.get("x-amz-target", "")
|
||||
return target.split(".")[-1]
|
||||
|
||||
@staticmethod
|
||||
def is_priced_job_start(httpx_response: httpx.Response) -> bool:
|
||||
return (
|
||||
TranscribePassthroughLoggingHandler._operation_from_response(httpx_response) == TRANSCRIBE_PRICED_OPERATION
|
||||
)
|
||||
|
||||
def schedule_priced_job_logging(
|
||||
self,
|
||||
httpx_response: httpx.Response,
|
||||
response_body: Mapping[str, object] | None,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
url_route: str,
|
||||
result: str,
|
||||
start_time: datetime,
|
||||
end_time: datetime,
|
||||
cache_hit: bool,
|
||||
request_body: Mapping[str, object],
|
||||
log: PassThroughLogDispatch,
|
||||
**kwargs: object, # kwargs-ok: the passthrough logging dispatch forwards shared logging kwargs to every handler
|
||||
) -> asyncio.Task[None]:
|
||||
task: Final = asyncio.create_task(
|
||||
self._price_then_log(
|
||||
httpx_response=httpx_response,
|
||||
started_job=started_transcription_job(response_body),
|
||||
logging_obj=logging_obj,
|
||||
url_route=url_route,
|
||||
result=result,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
cache_hit=cache_hit,
|
||||
request_body=request_body,
|
||||
log=log,
|
||||
**kwargs,
|
||||
)
|
||||
)
|
||||
self._pricing_tasks.add(task)
|
||||
task.add_done_callback(self._pricing_tasks.discard)
|
||||
return task
|
||||
|
||||
async def _price_then_log(
|
||||
self,
|
||||
httpx_response: httpx.Response,
|
||||
started_job: TranscriptionJobRecord | None,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
url_route: str,
|
||||
result: str,
|
||||
start_time: datetime,
|
||||
end_time: datetime,
|
||||
cache_hit: bool,
|
||||
request_body: Mapping[str, object],
|
||||
log: PassThroughLogDispatch,
|
||||
**kwargs: object, # kwargs-ok: the passthrough logging dispatch forwards shared logging kwargs to every handler
|
||||
) -> None:
|
||||
cost_per_second: Final = transcribe_cost_per_second()
|
||||
if cost_per_second is None:
|
||||
verbose_proxy_logger.error("%s left the model cost map, spend not recorded", TRANSCRIBE_PRICED_MODEL)
|
||||
return
|
||||
job_name: Final = request_body.get("TranscriptionJobName")
|
||||
aws_region_name: Final = httpx_response.request.url.host.split(".")[1]
|
||||
response_cost: Final = await self._job_pricer(
|
||||
job_name if isinstance(job_name, str) else "",
|
||||
aws_region_name,
|
||||
cost_per_second,
|
||||
started_job,
|
||||
)
|
||||
payload: Final = self.transcribe_passthrough_handler(
|
||||
httpx_response=httpx_response,
|
||||
logging_obj=logging_obj,
|
||||
url_route=url_route,
|
||||
result=result,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
cache_hit=cache_hit,
|
||||
request_body=request_body,
|
||||
response_cost=response_cost,
|
||||
**kwargs,
|
||||
)
|
||||
await log(
|
||||
logging_obj=logging_obj,
|
||||
standard_logging_response_object=payload["result"],
|
||||
result=result,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
cache_hit=cache_hit,
|
||||
**payload["kwargs"],
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def transcribe_passthrough_handler(
|
||||
httpx_response: httpx.Response,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
url_route: str,
|
||||
result: str,
|
||||
start_time: datetime,
|
||||
end_time: datetime,
|
||||
cache_hit: bool,
|
||||
request_body: Mapping[str, object],
|
||||
response_cost: float = 0.0,
|
||||
**kwargs: object, # kwargs-ok: the passthrough logging dispatch forwards shared logging kwargs to every handler
|
||||
) -> PassThroughEndpointLoggingTypedDict:
|
||||
try:
|
||||
operation: Final = TranscribePassthroughLoggingHandler._operation_from_response(httpx_response)
|
||||
model_name: Final = f"{TRANSCRIBE_CUSTOM_LLM_PROVIDER}/{operation}"
|
||||
|
||||
updated_kwargs: Final = { # mutable-ok: the logging pipeline requires a plain kwargs dict
|
||||
**kwargs,
|
||||
"model": model_name,
|
||||
"custom_llm_provider": TRANSCRIBE_CUSTOM_LLM_PROVIDER,
|
||||
"response_cost": response_cost,
|
||||
}
|
||||
logging_obj.model_call_details.update(
|
||||
model=model_name,
|
||||
custom_llm_provider=TRANSCRIBE_CUSTOM_LLM_PROVIDER,
|
||||
response_cost=response_cost,
|
||||
)
|
||||
|
||||
standard_logging_object: Final = get_standard_logging_object_payload(
|
||||
kwargs=updated_kwargs,
|
||||
init_response_obj=StandardPassThroughResponseObject(response=result),
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
logging_obj=logging_obj,
|
||||
status="success",
|
||||
)
|
||||
|
||||
handler_payload: Final[PassThroughEndpointLoggingTypedDict] = {
|
||||
"result": StandardPassThroughResponseObject(response=result),
|
||||
"kwargs": {**updated_kwargs, "standard_logging_object": standard_logging_object},
|
||||
}
|
||||
except Exception as e: # noqa: BLE001 # logging must never fail the forwarded request
|
||||
verbose_proxy_logger.exception("Error in Amazon Transcribe passthrough logging handler: %s", e)
|
||||
fallback_payload: Final[PassThroughEndpointLoggingTypedDict] = {
|
||||
"result": StandardPassThroughResponseObject(response=result),
|
||||
"kwargs": kwargs,
|
||||
}
|
||||
return fallback_payload
|
||||
return handler_payload
|
||||
|
|
@ -2669,17 +2669,22 @@ def _should_buffer_passthrough_response(response: httpx.Response) -> bool:
|
|||
"""
|
||||
Decide from the response headers whether the body must be read into memory.
|
||||
|
||||
JSON bodies (and upstream errors) stay buffered: spend logging, guardrails and
|
||||
managed-id rewriting inspect them, and they are small in practice. Everything
|
||||
else (jsonl batch results, octet-stream files, ...) is relayed to the client
|
||||
chunk by chunk so a large body is never resident in full (LIT-4009). A missing
|
||||
content-type is buffered because the body cannot be classified.
|
||||
JSON bodies (including the AWS JSON protocol media types) and upstream errors
|
||||
stay buffered: spend logging, guardrails and managed-id rewriting inspect them,
|
||||
and they are small in practice. Everything else (jsonl batch results,
|
||||
octet-stream files, ...) is relayed to the client chunk by chunk so a large
|
||||
body is never resident in full (LIT-4009). A missing content-type is buffered
|
||||
because the body cannot be classified.
|
||||
"""
|
||||
if response.status_code >= 400:
|
||||
return True
|
||||
content_type_header: Final[str] = response.headers.get("content-type", "")
|
||||
media_type: Final = content_type_header.split(";")[0].strip().lower()
|
||||
return media_type in ("", "application/json") or media_type.endswith("+json")
|
||||
return (
|
||||
media_type in ("", "application/json")
|
||||
or media_type.endswith("+json")
|
||||
or media_type.startswith("application/x-amz-json")
|
||||
)
|
||||
|
||||
|
||||
async def _relay_passthrough_response_bytes(
|
||||
|
|
|
|||
|
|
@ -28,6 +28,11 @@ from .llm_provider_handlers.cursor_passthrough_logging_handler import (
|
|||
from .llm_provider_handlers.gemini_passthrough_logging_handler import (
|
||||
GeminiPassthroughLoggingHandler,
|
||||
)
|
||||
from .llm_provider_handlers.transcribe_passthrough_logging_handler import (
|
||||
TRANSCRIBE_CUSTOM_LLM_PROVIDER,
|
||||
PassThroughLogDispatch,
|
||||
TranscribePassthroughLoggingHandler,
|
||||
)
|
||||
from .llm_provider_handlers.vertex_passthrough_logging_handler import (
|
||||
VertexPassthroughLoggingHandler,
|
||||
)
|
||||
|
|
@ -49,7 +54,15 @@ def _safe_response_text(httpx_response: httpx.Response) -> str:
|
|||
|
||||
|
||||
class PassThroughEndpointLogging:
|
||||
def __init__(self):
|
||||
def __init__(
|
||||
self,
|
||||
transcribe_handler: TranscribePassthroughLoggingHandler | None = None,
|
||||
log_dispatch: PassThroughLogDispatch | None = None,
|
||||
):
|
||||
self.transcribe_passthrough_logging_handler: Final = (
|
||||
transcribe_handler if transcribe_handler is not None else TranscribePassthroughLoggingHandler()
|
||||
)
|
||||
self._injected_log_dispatch: Final = log_dispatch
|
||||
self.TRACKED_VERTEX_METHOD_ROUTES = (
|
||||
"generateContent",
|
||||
"streamGenerateContent",
|
||||
|
|
@ -91,6 +104,10 @@ class PassThroughEndpointLogging:
|
|||
# Vertex AI Live API WebSocket
|
||||
self.TRACKED_VERTEX_AI_LIVE_ROUTES = ["/vertex_ai/live"]
|
||||
|
||||
@property
|
||||
def _log_dispatch(self) -> PassThroughLogDispatch:
|
||||
return self._injected_log_dispatch if self._injected_log_dispatch is not None else self._handle_logging
|
||||
|
||||
async def _handle_logging(
|
||||
self,
|
||||
logging_obj: LiteLLMLoggingObj,
|
||||
|
|
@ -257,6 +274,20 @@ class PassThroughEndpointLogging:
|
|||
)
|
||||
standard_logging_response_object = comprehend_medical_handler_result["result"] # rebind-ok: elif-chain
|
||||
kwargs = comprehend_medical_handler_result["kwargs"] # rebind-ok: elif-chain contract
|
||||
elif self.is_transcribe_route(custom_llm_provider):
|
||||
transcribe_handler_result: Final = TranscribePassthroughLoggingHandler.transcribe_passthrough_handler(
|
||||
httpx_response=httpx_response,
|
||||
logging_obj=logging_obj,
|
||||
url_route=url_route,
|
||||
result=result,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
cache_hit=cache_hit,
|
||||
request_body=request_body,
|
||||
**kwargs,
|
||||
)
|
||||
standard_logging_response_object = transcribe_handler_result["result"] # rebind-ok: elif-chain
|
||||
kwargs = transcribe_handler_result["kwargs"] # rebind-ok: elif-chain contract
|
||||
elif self.is_typesafe_route(custom_llm_provider):
|
||||
from .llm_provider_handlers.typesafe_passthrough_logging_handler import (
|
||||
TypeSafePassthroughLoggingHandler,
|
||||
|
|
@ -338,6 +369,24 @@ class PassThroughEndpointLogging:
|
|||
elif self.is_langfuse_route(url_route):
|
||||
# Don't log langfuse pass-through requests
|
||||
return
|
||||
elif self.is_transcribe_route(custom_llm_provider) and TranscribePassthroughLoggingHandler.is_priced_job_start(
|
||||
httpx_response
|
||||
):
|
||||
self.transcribe_passthrough_logging_handler.schedule_priced_job_logging(
|
||||
httpx_response=httpx_response,
|
||||
response_body=response_body if isinstance(response_body, dict) else None,
|
||||
logging_obj=logging_obj,
|
||||
url_route=url_route,
|
||||
result=result,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
cache_hit=cache_hit,
|
||||
request_body=request_body,
|
||||
log=self._log_dispatch,
|
||||
standard_pass_through_logging_payload=passthrough_logging_payload,
|
||||
**kwargs,
|
||||
)
|
||||
return
|
||||
else:
|
||||
normalized_llm_passthrough_logging_payload: Final = self.normalize_llm_passthrough_logging_payload(
|
||||
httpx_response=httpx_response,
|
||||
|
|
@ -367,7 +416,7 @@ class PassThroughEndpointLogging:
|
|||
kwargs=kwargs,
|
||||
)
|
||||
|
||||
await self._handle_logging(
|
||||
await self._log_dispatch(
|
||||
logging_obj=logging_obj,
|
||||
standard_logging_response_object=standard_logging_response_object,
|
||||
result=result,
|
||||
|
|
@ -409,6 +458,9 @@ class PassThroughEndpointLogging:
|
|||
def is_comprehend_medical_route(self, custom_llm_provider: str | None) -> bool:
|
||||
return custom_llm_provider == "comprehendmedical"
|
||||
|
||||
def is_transcribe_route(self, custom_llm_provider: str | None) -> bool:
|
||||
return custom_llm_provider == TRANSCRIBE_CUSTOM_LLM_PROVIDER
|
||||
|
||||
def is_typesafe_route(self, custom_llm_provider: str | None) -> bool:
|
||||
return custom_llm_provider == "typesafe"
|
||||
|
||||
|
|
|
|||
|
|
@ -17128,6 +17128,7 @@ _GENERAL_SETTINGS_CONFIG_LIST_FIELD_TYPES: Final[Mapping[str, str]] = MappingPro
|
|||
"disable_auto_add_proxy_admin_to_teams": "Boolean",
|
||||
"apply_user_budget_to_team_keys": "Boolean",
|
||||
"user_api_key_cache_max_size": "Integer",
|
||||
"transcribe_media_buckets": "List",
|
||||
}
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -46324,6 +46324,16 @@
|
|||
"/v1/audio/speech"
|
||||
]
|
||||
},
|
||||
"transcribe/StartTranscriptionJob": {
|
||||
"input_cost_per_second": 0.0001,
|
||||
"litellm_provider": "transcribe",
|
||||
"mode": "audio_transcription",
|
||||
"output_cost_per_second": 0.0,
|
||||
"source": "https://aws.amazon.com/transcribe/pricing/",
|
||||
"metadata": {
|
||||
"notes": "Amazon Transcribe standard batch transcription, billed per second of audio with no minimum. Same rate in every region of the AWS Price List offer file for transcribe (checked 2026-09-17)"
|
||||
}
|
||||
},
|
||||
"aws_polly/standard": {
|
||||
"input_cost_per_character": 4e-06,
|
||||
"litellm_provider": "aws_polly",
|
||||
|
|
|
|||
|
|
@ -86,7 +86,7 @@ locals {
|
|||
"/queue/chat/*",
|
||||
"/v1beta/*",
|
||||
"/interactions/*",
|
||||
"/anthropic/*", "/azure/*", "/azure_ai/*", "/aws/*", "/bedrock/*", "/comprehendmedical*",
|
||||
"/anthropic/*", "/azure/*", "/azure_ai/*", "/aws/*", "/bedrock/*", "/comprehendmedical*", "/transcribe*",
|
||||
"/cohere/*", "/gemini/*", "/google/*",
|
||||
"/vertex_ai/*", "/vertex-ai/*",
|
||||
"/assemblyai/*", "/eu.assemblyai/*",
|
||||
|
|
|
|||
|
|
@ -55,7 +55,7 @@ locals {
|
|||
"/queue/chat/*",
|
||||
"/v1beta/*",
|
||||
"/interactions/*",
|
||||
"/anthropic/*", "/azure/*", "/azure_ai/*", "/aws/*", "/bedrock/*", "/comprehendmedical*",
|
||||
"/anthropic/*", "/azure/*", "/azure_ai/*", "/aws/*", "/bedrock/*", "/comprehendmedical*", "/transcribe*",
|
||||
"/cohere/*", "/gemini/*", "/google/*",
|
||||
"/vertex_ai/*", "/vertex-ai/*",
|
||||
"/assemblyai/*", "/eu.assemblyai/*",
|
||||
|
|
|
|||
|
|
@ -411,6 +411,8 @@ async def test_pass_through_request_logging_failure_with_stream(
|
|||
PROTOCOL_CONSTRAINED_PASS_THROUGH_ROUTES = {
|
||||
"/comprehendmedical": {"POST"},
|
||||
"/comprehendmedical/{operation}": {"POST"},
|
||||
"/transcribe": {"POST"},
|
||||
"/transcribe/{operation}": {"POST"},
|
||||
}
|
||||
|
||||
|
||||
|
|
@ -418,9 +420,7 @@ def test_pass_through_routes_support_all_methods():
|
|||
"""
|
||||
A pass-through route fronts a whole provider API, so narrowing its method
|
||||
set turns a request the upstream would have accepted into a 405. The
|
||||
exceptions are providers whose wire protocol admits only one method: Amazon
|
||||
Comprehend Medical speaks AWS JSON 1.1, which is POST-only, so there is no
|
||||
other method to forward.
|
||||
exceptions are the POST-only protocol routes listed above.
|
||||
"""
|
||||
from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
|
||||
router as llm_router,
|
||||
|
|
|
|||
|
|
@ -116,6 +116,8 @@ def test_is_pure_asgi_not_base_http_middleware():
|
|||
# Bare AWS-SDK-shaped route carries the operation in X-Amz-Target and writes SpendLogs
|
||||
("/comprehendmedical", (BillableCategory.LLM, "/comprehendmedical")),
|
||||
("/comprehendmedical/DetectEntitiesV2", (BillableCategory.LLM, "/comprehendmedical")),
|
||||
("/transcribe", (BillableCategory.LLM, "/transcribe")),
|
||||
("/transcribe/StartTranscriptionJob", (BillableCategory.LLM, "/transcribe")),
|
||||
("/mcp", (BillableCategory.MCP, "/mcp")),
|
||||
("/mcp/", (BillableCategory.MCP, "/mcp")),
|
||||
("/mcp/tools/list", (BillableCategory.MCP, "/mcp")),
|
||||
|
|
|
|||
|
|
@ -0,0 +1,942 @@
|
|||
import asyncio
|
||||
import io
|
||||
import json
|
||||
import wave
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
|
||||
from litellm.proxy.pass_through_endpoints.llm_provider_handlers.transcribe_passthrough_logging_handler import (
|
||||
TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS,
|
||||
TRANSCRIBE_OWNER_TAG,
|
||||
TranscribePassthroughLoggingHandler,
|
||||
TranscribeRefusal,
|
||||
TranscriptionJobRecord,
|
||||
media_file_seconds,
|
||||
media_predates_job,
|
||||
price_transcription_job,
|
||||
requested_media_format,
|
||||
s3_media_url,
|
||||
started_transcription_job,
|
||||
transcribe_admin_only_refusal,
|
||||
transcribe_cost_per_second,
|
||||
transcribe_job_access_refusal,
|
||||
transcribe_media_buckets,
|
||||
transcribe_owned_start_request,
|
||||
transcribe_storage_refusal,
|
||||
transcribe_supported_operations,
|
||||
transcribe_unpriceable_request_reason,
|
||||
write_media_within_limit,
|
||||
)
|
||||
from litellm.proxy.pass_through_endpoints.success_handler import (
|
||||
PassThroughEndpointLogging,
|
||||
)
|
||||
|
||||
COST_PER_SECOND = 0.0001
|
||||
|
||||
|
||||
def _make_response(operation: str) -> httpx.Response:
|
||||
request = httpx.Request(
|
||||
"POST",
|
||||
"https://transcribe.us-west-2.amazonaws.com/",
|
||||
headers={"X-Amz-Target": f"Transcribe.{operation}"},
|
||||
)
|
||||
return httpx.Response(200, request=request, text='{"TranscriptionJob": {}}')
|
||||
|
||||
|
||||
async def _relayed_response(operation: str, body: bytes) -> httpx.Response:
|
||||
response = httpx.Response(
|
||||
200,
|
||||
request=_make_response(operation).request,
|
||||
headers={"content-type": "application/x-amz-json-1.1"},
|
||||
stream=httpx.ByteStream(body),
|
||||
)
|
||||
async for _ in response.aiter_bytes():
|
||||
pass
|
||||
await response.aclose()
|
||||
return response
|
||||
|
||||
|
||||
def _make_logging_obj() -> MagicMock:
|
||||
logging_obj = MagicMock()
|
||||
logging_obj.litellm_call_id = "test-call-id"
|
||||
logging_obj.model_call_details = {}
|
||||
return logging_obj
|
||||
|
||||
|
||||
async def _no_sleep(_: float) -> None:
|
||||
return None
|
||||
|
||||
|
||||
MEDIA_URI = "s3://b/a.wav"
|
||||
CREATED_AT = 1_789_682_363.696
|
||||
|
||||
|
||||
def _job(
|
||||
status: str, media_uri: str | None = MEDIA_URI, created_at: float | None = CREATED_AT, **members: object
|
||||
) -> dict[str, object]:
|
||||
media = {"Media": {"MediaFileUri": media_uri}} if media_uri else {}
|
||||
created = {"CreationTime": created_at} if created_at is not None else {}
|
||||
return {"TranscriptionJob": {"TranscriptionJobStatus": status, **media, **created, **members}}
|
||||
|
||||
|
||||
async def _no_media(uri: str, created_at: float) -> float | None:
|
||||
raise AssertionError("the media must not be measured on this path")
|
||||
|
||||
|
||||
def _media_probe(*durations: float | None | Exception):
|
||||
remaining = list(durations)
|
||||
measured: list[tuple[str, float]] = []
|
||||
|
||||
async def media_seconds(uri: str, created_at: float) -> float | None:
|
||||
measured.append((uri, created_at))
|
||||
outcome = remaining.pop(0) if len(remaining) > 1 else remaining[0]
|
||||
if isinstance(outcome, Exception):
|
||||
raise outcome
|
||||
return outcome
|
||||
|
||||
return media_seconds, measured
|
||||
|
||||
|
||||
def _sequence(*jobs: dict[str, object]):
|
||||
remaining = list(jobs)
|
||||
seen: list[str] = []
|
||||
|
||||
async def get_job(job_name: str) -> dict[str, object]:
|
||||
seen.append(job_name)
|
||||
return remaining.pop(0) if len(remaining) > 1 else remaining[0]
|
||||
|
||||
return get_job, seen
|
||||
|
||||
|
||||
def _aws_error(error_type: str) -> httpx.HTTPStatusError:
|
||||
request = httpx.Request("POST", "https://transcribe.us-west-2.amazonaws.com/")
|
||||
response = httpx.Response(400, request=request, json={"__type": error_type, "message": "nope"})
|
||||
return httpx.HTTPStatusError("400", request=request, response=response)
|
||||
|
||||
|
||||
def _missing_job(error_type: str):
|
||||
seen: list[str] = []
|
||||
|
||||
async def get_job(job_name: str) -> dict[str, object]:
|
||||
seen.append(job_name)
|
||||
raise _aws_error(error_type)
|
||||
|
||||
return get_job, seen
|
||||
|
||||
|
||||
class TestTranscribeSupportedOperations:
|
||||
def test_matches_the_installed_botocore_service_model(self):
|
||||
from botocore.session import get_session
|
||||
|
||||
assert transcribe_supported_operations() == frozenset(
|
||||
get_session().get_service_model("transcribe").operation_names
|
||||
)
|
||||
|
||||
|
||||
class TestTranscribeCostMap:
|
||||
def test_start_transcription_job_is_priced_per_second_of_audio(self):
|
||||
entry = litellm.model_cost["transcribe/StartTranscriptionJob"]
|
||||
|
||||
assert entry["litellm_provider"] == "transcribe"
|
||||
assert entry["mode"] == "audio_transcription"
|
||||
assert transcribe_cost_per_second() == entry["input_cost_per_second"] > 0
|
||||
|
||||
def test_missing_or_malformed_entry_yields_no_rate(self, monkeypatch: pytest.MonkeyPatch):
|
||||
monkeypatch.setitem(litellm.model_cost, "transcribe/StartTranscriptionJob", {"input_cost_per_second": "x"})
|
||||
assert transcribe_cost_per_second() is None
|
||||
monkeypatch.delitem(litellm.model_cost, "transcribe/StartTranscriptionJob")
|
||||
assert transcribe_cost_per_second() is None
|
||||
|
||||
|
||||
class TestTranscribeUnpriceableRequestReason:
|
||||
def test_plain_start_transcription_job_is_allowed(self):
|
||||
body = {"TranscriptionJobName": "j", "Media": {"MediaFileUri": MEDIA_URI}}
|
||||
assert transcribe_unpriceable_request_reason("StartTranscriptionJob", body, COST_PER_SECOND) is None
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"body",
|
||||
[
|
||||
{"Media": {"MediaFileUri": "s3://b/a.mp4"}},
|
||||
{"Media": {"MediaFileUri": "s3://b/a.wav"}, "MediaFormat": "webm"},
|
||||
{"Media": {"MediaFileUri": "s3://b/recording"}},
|
||||
{"TranscriptionJobName": "j"},
|
||||
],
|
||||
)
|
||||
def test_media_whose_length_cannot_be_read_is_rejected(self, body: dict[str, object]):
|
||||
reason = transcribe_unpriceable_request_reason("StartTranscriptionJob", body, COST_PER_SECOND)
|
||||
assert reason is not None and "MediaFormat" in reason
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"body",
|
||||
[
|
||||
{"Media": {"MediaFileUri": "s3://b/a.mp4"}, "MediaFormat": "mp3"},
|
||||
{"Media": {"MediaFileUri": "https://s3.us-west-2.amazonaws.com/b/a.FLAC?x=1"}},
|
||||
{"Media": {"MediaFileUri": "s3://b/dir.v2/a.ogg"}},
|
||||
],
|
||||
)
|
||||
def test_measurable_media_is_allowed(self, body: dict[str, object]):
|
||||
assert transcribe_unpriceable_request_reason("StartTranscriptionJob", body, COST_PER_SECOND) is None
|
||||
|
||||
def test_read_only_operations_are_allowed_without_a_rate(self):
|
||||
assert transcribe_unpriceable_request_reason("GetTranscriptionJob", {}, None) is None
|
||||
assert transcribe_unpriceable_request_reason("ListTranscriptionJobs", {}, None) is None
|
||||
|
||||
def test_start_transcription_job_needs_a_rate(self):
|
||||
reason = transcribe_unpriceable_request_reason("StartTranscriptionJob", {"TranscriptionJobName": "j"}, None)
|
||||
assert reason is not None and "model cost map" in reason
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"operation", ["StartCallAnalyticsJob", "StartMedicalScribeJob", "StartMedicalTranscriptionJob"]
|
||||
)
|
||||
def test_unpriced_job_classes_are_rejected(self, operation: str):
|
||||
reason = transcribe_unpriceable_request_reason(operation, {}, COST_PER_SECOND)
|
||||
assert reason is not None and operation in reason
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("body", "member"),
|
||||
[
|
||||
({"ContentRedaction": {"RedactionType": "PII", "RedactionOutput": "redacted"}}, "ContentRedaction"),
|
||||
({"ToxicityDetection": [{"ToxicityCategories": ["ALL"]}]}, "ToxicityDetection"),
|
||||
({"ModelSettings": {"LanguageModelName": "clm"}}, "ModelSettings.LanguageModelName"),
|
||||
(
|
||||
{
|
||||
"IdentifyLanguage": True,
|
||||
"LanguageIdSettings": {"en-US": {"VocabularyName": "v"}, "fr-FR": {"LanguageModelName": "clm"}},
|
||||
},
|
||||
"LanguageIdSettings.fr-FR.LanguageModelName",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_surcharged_features_are_rejected(self, body: dict[str, object], member: str):
|
||||
reason = transcribe_unpriceable_request_reason(
|
||||
"StartTranscriptionJob", {**body, "Media": {"MediaFileUri": MEDIA_URI}}, COST_PER_SECOND
|
||||
)
|
||||
assert reason is not None and member in reason
|
||||
|
||||
def test_settings_without_a_custom_model_are_allowed(self):
|
||||
body = {
|
||||
"ModelSettings": {},
|
||||
"LanguageIdSettings": {"en-US": {"VocabularyName": "v"}},
|
||||
"Media": {"MediaFileUri": MEDIA_URI},
|
||||
}
|
||||
assert transcribe_unpriceable_request_reason("StartTranscriptionJob", body, COST_PER_SECOND) is None
|
||||
|
||||
|
||||
class TestRequestedMediaFormat:
|
||||
def test_explicit_media_format_wins_over_the_extension(self):
|
||||
assert requested_media_format({"MediaFormat": "MP3", "Media": {"MediaFileUri": "s3://b/a.wav"}}) == "mp3"
|
||||
|
||||
def test_extension_is_read_from_the_uri_path_only(self):
|
||||
assert requested_media_format({"Media": {"MediaFileUri": "https://h/b/a.wav?sig=x.y"}}) == "wav"
|
||||
assert requested_media_format({"Media": {"MediaFileUri": "s3://b.name/a"}}) is None
|
||||
assert requested_media_format({"Media": {"MediaFileUri": 7}}) is None
|
||||
|
||||
|
||||
class TestS3MediaUrl:
|
||||
def test_s3_uri_maps_to_the_regional_virtual_hosted_endpoint(self):
|
||||
assert (
|
||||
s3_media_url("s3://my-bucket/dir/a b.wav", "us-west-2")
|
||||
== "https://my-bucket.s3.us-west-2.amazonaws.com/dir/a%20b.wav"
|
||||
)
|
||||
|
||||
def test_dotted_bucket_maps_to_the_regional_path_style_endpoint(self):
|
||||
assert (
|
||||
s3_media_url("s3://media.example.com/dir/a b.wav", "us-west-2")
|
||||
== "https://s3.us-west-2.amazonaws.com/media.example.com/dir/a%20b.wav"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"media_uri",
|
||||
[
|
||||
"https://evil.example.com/a.wav",
|
||||
"https://my-bucket.s3.us-west-2.amazonaws.com@evil.example.com/a.wav",
|
||||
"https://amazonaws.com/a.wav",
|
||||
"http://my-bucket.s3.us-west-2.amazonaws.com/a.wav",
|
||||
],
|
||||
)
|
||||
def test_hosts_outside_the_aws_partition_or_off_https_are_never_signed_for(self, media_uri: str):
|
||||
assert s3_media_url(media_uri, "us-west-2") is None
|
||||
|
||||
def test_https_uri_is_used_as_given(self):
|
||||
assert (
|
||||
s3_media_url("https://my-bucket.s3.eu-west-1.amazonaws.com/a.wav", "us-west-2")
|
||||
== "https://my-bucket.s3.eu-west-1.amazonaws.com/a.wav"
|
||||
)
|
||||
|
||||
|
||||
class _ChunkedStream(httpx.AsyncByteStream):
|
||||
def __init__(self, *chunks: bytes) -> None:
|
||||
self._chunks = chunks
|
||||
|
||||
async def __aiter__(self):
|
||||
for chunk in self._chunks:
|
||||
yield chunk
|
||||
|
||||
|
||||
def _media_response(*chunks: bytes, content_length: int | None) -> httpx.Response:
|
||||
headers = {"content-length": str(content_length)} if content_length is not None else {}
|
||||
return httpx.Response(200, headers=headers, stream=_ChunkedStream(*chunks))
|
||||
|
||||
|
||||
class TestWriteMediaWithinLimit:
|
||||
@pytest.mark.asyncio
|
||||
async def test_media_within_the_cap_is_written_whole(self):
|
||||
media_file = io.BytesIO()
|
||||
assert await write_media_within_limit(_media_response(b"abc", b"def", content_length=6), media_file, 6) is True
|
||||
assert media_file.getvalue() == b"abcdef"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_advertised_size_over_the_cap_is_refused_before_downloading(self):
|
||||
media_file = io.BytesIO()
|
||||
assert await write_media_within_limit(_media_response(b"abcdef", content_length=7), media_file, 6) is False
|
||||
assert media_file.getvalue() == b""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_growing_past_the_cap_is_cut_off(self):
|
||||
media_file = io.BytesIO()
|
||||
response = _media_response(b"abc", b"def", b"ghi", content_length=None)
|
||||
assert await write_media_within_limit(response, media_file, 5) is False
|
||||
assert media_file.getvalue() == b"abcdef"
|
||||
|
||||
|
||||
class TestPriceTranscriptionJob:
|
||||
@pytest.mark.asyncio
|
||||
async def test_polls_until_completed_then_charges_the_media_length_rounded_up(self):
|
||||
get_job, seen = _sequence(_job("IN_PROGRESS"), _job("IN_PROGRESS"), _job("COMPLETED"))
|
||||
media_seconds, measured = _media_probe(17.577)
|
||||
|
||||
cost = await price_transcription_job("job-1", COST_PER_SECOND, get_job, media_seconds, sleep=_no_sleep)
|
||||
|
||||
assert cost == pytest.approx(18 * COST_PER_SECOND)
|
||||
assert seen == ["job-1", "job-1", "job-1"]
|
||||
assert measured == [(MEDIA_URI, CREATED_AT)]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_failed_poll_is_retried_instead_of_ending_pricing(self):
|
||||
remaining = [httpx.ConnectError("aws blip"), None]
|
||||
|
||||
async def get_job(job_name: str) -> dict[str, object]:
|
||||
outcome = remaining.pop(0)
|
||||
if outcome is not None:
|
||||
raise outcome
|
||||
return _job("COMPLETED")
|
||||
|
||||
media_seconds, _ = _media_probe(3.0)
|
||||
|
||||
cost = await price_transcription_job("job-1", COST_PER_SECOND, get_job, media_seconds, sleep=_no_sleep)
|
||||
|
||||
assert cost == pytest.approx(3 * COST_PER_SECOND)
|
||||
assert remaining == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failed_job_costs_nothing(self):
|
||||
get_job, _ = _sequence(_job("FAILED"))
|
||||
|
||||
assert await price_transcription_job("job-1", COST_PER_SECOND, get_job, _no_media, sleep=_no_sleep) == 0.0
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_job_deleted_before_it_is_polled_is_charged_for_the_media_it_was_started_with(self):
|
||||
get_job, seen = _missing_job("BadRequestException")
|
||||
media_seconds, measured = _media_probe(17.577)
|
||||
started = started_transcription_job(
|
||||
{"TranscriptionJob": {"Media": {"MediaFileUri": "s3://b/started.wav"}, "CreationTime": 5.0}}
|
||||
)
|
||||
|
||||
cost = await price_transcription_job(
|
||||
"job-1", COST_PER_SECOND, get_job, media_seconds, sleep=_no_sleep, started_job=started
|
||||
)
|
||||
|
||||
assert cost == pytest.approx(18 * COST_PER_SECOND)
|
||||
assert seen == ["job-1"]
|
||||
assert measured == [("s3://b/started.wav", 5.0)]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_job_not_found_by_transcribe_is_charged_the_maximum_without_a_start_record(self):
|
||||
get_job, seen = _missing_job("com.amazonaws.transcribe#NotFoundException")
|
||||
|
||||
cost = await price_transcription_job("job-1", COST_PER_SECOND, get_job, _no_media, sleep=_no_sleep)
|
||||
|
||||
assert cost == pytest.approx(TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS * COST_PER_SECOND)
|
||||
assert seen == ["job-1"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_throttled_poll_is_retried_rather_than_treated_as_a_missing_job(self):
|
||||
remaining = ["LimitExceededException", None]
|
||||
|
||||
async def get_job(job_name: str) -> dict[str, object]:
|
||||
error_type = remaining.pop(0)
|
||||
if error_type is not None:
|
||||
raise _aws_error(error_type)
|
||||
return _job("COMPLETED")
|
||||
|
||||
media_seconds, _ = _media_probe(3.0)
|
||||
|
||||
cost = await price_transcription_job("job-1", COST_PER_SECOND, get_job, media_seconds, sleep=_no_sleep)
|
||||
|
||||
assert cost == pytest.approx(3 * COST_PER_SECOND)
|
||||
assert remaining == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_job_that_never_finishes_is_charged_the_maximum(self):
|
||||
get_job, seen = _sequence(_job("IN_PROGRESS"))
|
||||
|
||||
cost = await price_transcription_job(
|
||||
"job-1", COST_PER_SECOND, get_job, _no_media, sleep=_no_sleep, max_attempts=3
|
||||
)
|
||||
|
||||
assert cost == pytest.approx(TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS * COST_PER_SECOND)
|
||||
assert len(seen) == 3
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_media_that_cannot_be_read_is_charged_the_maximum(self):
|
||||
get_job, _ = _sequence(_job("COMPLETED"))
|
||||
media_seconds, measured = _media_probe(None)
|
||||
|
||||
cost = await price_transcription_job("job-1", COST_PER_SECOND, get_job, media_seconds, sleep=_no_sleep)
|
||||
|
||||
assert cost == pytest.approx(TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS * COST_PER_SECOND)
|
||||
assert measured == [(MEDIA_URI, CREATED_AT)]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_media_fetch_is_retried_then_charged_the_maximum(self):
|
||||
get_job, _ = _sequence(_job("COMPLETED"))
|
||||
media_seconds, measured = _media_probe(httpx.ReadTimeout("s3 slow"))
|
||||
|
||||
cost = await price_transcription_job("job-1", COST_PER_SECOND, get_job, media_seconds, sleep=_no_sleep)
|
||||
|
||||
assert cost == pytest.approx(TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS * COST_PER_SECOND)
|
||||
assert len(measured) == 3
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_media_fetch_recovers_after_a_transient_failure(self):
|
||||
get_job, _ = _sequence(_job("COMPLETED"))
|
||||
media_seconds, measured = _media_probe(httpx.ReadTimeout("s3 slow"), 60.0)
|
||||
|
||||
cost = await price_transcription_job("job-1", COST_PER_SECOND, get_job, media_seconds, sleep=_no_sleep)
|
||||
|
||||
assert cost == pytest.approx(60 * COST_PER_SECOND)
|
||||
assert len(measured) == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_completed_job_without_media_uri_is_charged_the_maximum(self):
|
||||
get_job, _ = _sequence(_job("COMPLETED", media_uri=None))
|
||||
|
||||
cost = await price_transcription_job("job-1", COST_PER_SECOND, get_job, _no_media, sleep=_no_sleep)
|
||||
|
||||
assert cost == pytest.approx(TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS * COST_PER_SECOND)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_completed_job_without_creation_time_is_charged_the_maximum_unmeasured(self):
|
||||
get_job, _ = _sequence(_job("COMPLETED", created_at=None))
|
||||
media_seconds, measured = _media_probe(60.0)
|
||||
|
||||
cost = await price_transcription_job("job-1", COST_PER_SECOND, get_job, media_seconds, sleep=_no_sleep)
|
||||
|
||||
assert cost == pytest.approx(TRANSCRIBE_MAX_MEDIA_DURATION_SECONDS * COST_PER_SECOND)
|
||||
assert measured == []
|
||||
|
||||
|
||||
class TestMediaFileSeconds:
|
||||
def test_reads_the_duration_from_the_file_on_disk(self, tmp_path: Path):
|
||||
media = tmp_path / "a.wav"
|
||||
with wave.open(str(media), "wb") as out:
|
||||
out.setnchannels(1)
|
||||
out.setsampwidth(2)
|
||||
out.setframerate(8000)
|
||||
out.writeframes(bytes(2 * 12_000))
|
||||
|
||||
assert media_file_seconds(media) == pytest.approx(1.5)
|
||||
|
||||
def test_undecodable_media_yields_no_duration(self, tmp_path: Path):
|
||||
media = tmp_path / "a.wav"
|
||||
_ = media.write_bytes(b"not audio at all")
|
||||
|
||||
assert media_file_seconds(media) is None
|
||||
|
||||
|
||||
class TestStartedTranscriptionJob:
|
||||
def test_reads_the_media_and_creation_time_from_the_start_response(self):
|
||||
started = started_transcription_job(
|
||||
{
|
||||
"TranscriptionJob": {
|
||||
"TranscriptionJobName": "j",
|
||||
"Media": {"MediaFileUri": "s3://b/a.wav"},
|
||||
"CreationTime": 1.5,
|
||||
"TranscriptionJobStatus": "IN_PROGRESS",
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
assert started == TranscriptionJobRecord(
|
||||
TranscriptionJobStatus="IN_PROGRESS", CreationTime=1.5, Media={"MediaFileUri": "s3://b/a.wav"}
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("body", [None, {"Message": "throttled"}, {"TranscriptionJob": {"CreationTime": "soon"}}])
|
||||
def test_unreadable_start_response_yields_no_record(self, body: dict[str, object] | None):
|
||||
assert started_transcription_job(body) is None
|
||||
|
||||
|
||||
class TestMediaPredatesJob:
|
||||
LAST_MODIFIED = "Thu, 17 Sep 2026 17:45:00 GMT"
|
||||
LAST_MODIFIED_EPOCH = 1_789_667_100.0
|
||||
|
||||
def test_object_written_before_the_job_counts(self):
|
||||
assert media_predates_job(httpx.Headers({"Last-Modified": self.LAST_MODIFIED}), self.LAST_MODIFIED_EPOCH + 30)
|
||||
|
||||
def test_object_written_in_the_same_second_as_the_job_counts(self):
|
||||
assert media_predates_job(httpx.Headers({"Last-Modified": self.LAST_MODIFIED}), self.LAST_MODIFIED_EPOCH - 0.4)
|
||||
|
||||
def test_object_rewritten_after_the_job_does_not_count(self):
|
||||
assert not media_predates_job(
|
||||
httpx.Headers({"Last-Modified": self.LAST_MODIFIED}), self.LAST_MODIFIED_EPOCH - 30
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("headers", [{}, {"Last-Modified": "yesterday"}])
|
||||
def test_unknown_modification_time_does_not_count(self, headers: dict[str, str]):
|
||||
assert not media_predates_job(httpx.Headers(headers), self.LAST_MODIFIED_EPOCH + 30)
|
||||
|
||||
|
||||
VIRTUAL_KEY = UserAPIKeyAuth(api_key="hashed-key-a", user_id="user-a", team_id="team-a")
|
||||
OTHER_VIRTUAL_KEY = UserAPIKeyAuth(api_key="hashed-key-b", user_id="user-b", team_id="team-b")
|
||||
ADMIN_KEY = UserAPIKeyAuth(api_key="hashed-admin", user_role=LitellmUserRoles.PROXY_ADMIN)
|
||||
|
||||
|
||||
class TestTranscribeAdminOnlyRefusal:
|
||||
@pytest.mark.parametrize("operation", ["StartTranscriptionJob", "GetTranscriptionJob", "DeleteTranscriptionJob"])
|
||||
def test_job_scoped_operations_are_open_to_virtual_keys(self, operation: str):
|
||||
assert transcribe_admin_only_refusal(operation, VIRTUAL_KEY) is None
|
||||
|
||||
@pytest.mark.parametrize("operation", ["ListTranscriptionJobs", "ListVocabularies", "DeleteVocabulary"])
|
||||
def test_account_wide_operations_are_refused_for_virtual_keys(self, operation: str):
|
||||
refusal = transcribe_admin_only_refusal(operation, VIRTUAL_KEY)
|
||||
|
||||
assert refusal is not None
|
||||
assert refusal.status_code == 403
|
||||
assert operation in refusal.detail
|
||||
|
||||
@pytest.mark.parametrize("operation", ["ListTranscriptionJobs", "DeleteVocabulary"])
|
||||
def test_account_wide_operations_are_open_to_proxy_admins(self, operation: str):
|
||||
assert transcribe_admin_only_refusal(operation, ADMIN_KEY) is None
|
||||
|
||||
|
||||
ALLOWED_BUCKETS = frozenset({"tenant-media", "tenant-transcripts"})
|
||||
|
||||
|
||||
def _start_body(media_uri: str = "s3://tenant-media/call.wav", **members: object) -> dict[str, object]:
|
||||
return {"TranscriptionJobName": "j", "Media": {"MediaFileUri": media_uri}, **members}
|
||||
|
||||
|
||||
class TestTranscribeMediaBuckets:
|
||||
def test_a_list_of_bucket_names_is_read_from_general_settings(self):
|
||||
assert transcribe_media_buckets({"transcribe_media_buckets": ["a", "b"]}) == frozenset({"a", "b"})
|
||||
|
||||
@pytest.mark.parametrize("settings", [{}, {"transcribe_media_buckets": "a"}, {"transcribe_media_buckets": [1]}])
|
||||
def test_a_missing_or_malformed_setting_reads_as_unset(self, settings: dict[str, object]):
|
||||
assert transcribe_media_buckets(settings) is None
|
||||
|
||||
|
||||
class TestTranscribeStorageRefusal:
|
||||
def test_media_and_output_in_listed_buckets_are_allowed(self):
|
||||
body = _start_body(OutputBucketName="tenant-transcripts", OutputKey="out/")
|
||||
|
||||
assert transcribe_storage_refusal(body, ALLOWED_BUCKETS, VIRTUAL_KEY) is None
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"media_uri",
|
||||
[
|
||||
"s3://other-tenant/call.wav",
|
||||
"https://tenant-media.s3.us-west-2.amazonaws.com/call.wav",
|
||||
"s3://",
|
||||
],
|
||||
)
|
||||
def test_media_outside_the_listed_buckets_is_refused(self, media_uri: str):
|
||||
refusal = transcribe_storage_refusal(_start_body(media_uri), ALLOWED_BUCKETS, VIRTUAL_KEY)
|
||||
|
||||
assert refusal is not None
|
||||
assert refusal.status_code == 403
|
||||
assert "Media.MediaFileUri" in refusal.detail
|
||||
|
||||
def test_redacted_media_outside_the_listed_buckets_is_refused(self):
|
||||
body = {
|
||||
"TranscriptionJobName": "j",
|
||||
"Media": {"MediaFileUri": "s3://tenant-media/call.wav", "RedactedMediaFileUri": "s3://other-tenant/c.wav"},
|
||||
}
|
||||
|
||||
refusal = transcribe_storage_refusal(body, ALLOWED_BUCKETS, VIRTUAL_KEY)
|
||||
|
||||
assert refusal is not None
|
||||
assert "Media.RedactedMediaFileUri" in refusal.detail
|
||||
|
||||
@pytest.mark.parametrize("output", ["other-tenant", 7])
|
||||
def test_an_output_bucket_outside_the_listed_buckets_is_refused(self, output: object):
|
||||
refusal = transcribe_storage_refusal(_start_body(OutputBucketName=output), ALLOWED_BUCKETS, VIRTUAL_KEY)
|
||||
|
||||
assert refusal is not None
|
||||
assert refusal.status_code == 403
|
||||
assert "OutputBucketName" in refusal.detail
|
||||
|
||||
@pytest.mark.parametrize("member", ["DataAccessRoleArn", "JobExecutionSettings"])
|
||||
def test_a_caller_chosen_role_is_refused(self, member: str):
|
||||
refusal = transcribe_storage_refusal(_start_body(**{member: "x"}), ALLOWED_BUCKETS, VIRTUAL_KEY)
|
||||
|
||||
assert refusal is not None
|
||||
assert refusal.status_code == 403
|
||||
assert member in refusal.detail
|
||||
|
||||
def test_an_unset_bucket_list_refuses_virtual_keys(self):
|
||||
refusal = transcribe_storage_refusal(_start_body(), None, VIRTUAL_KEY)
|
||||
|
||||
assert refusal is not None
|
||||
assert refusal.status_code == 403
|
||||
assert "transcribe_media_buckets" in refusal.detail
|
||||
|
||||
@pytest.mark.parametrize("allowed", [None, ALLOWED_BUCKETS])
|
||||
def test_proxy_admins_are_not_restricted(self, allowed: frozenset[str] | None):
|
||||
body = _start_body("s3://other-tenant/call.wav", DataAccessRoleArn="arn:aws:iam::1:role/r")
|
||||
|
||||
assert transcribe_storage_refusal(body, allowed, ADMIN_KEY) is None
|
||||
|
||||
|
||||
class TestTranscribeOwnedStartRequest:
|
||||
def test_the_caller_identity_is_appended_to_the_job_tags(self):
|
||||
body = {"TranscriptionJobName": "j", "Tags": [{"Key": "env", "Value": "qa"}]}
|
||||
|
||||
owned = transcribe_owned_start_request(body, VIRTUAL_KEY)
|
||||
|
||||
assert owned == {
|
||||
"TranscriptionJobName": "j",
|
||||
"Tags": ({"Key": "env", "Value": "qa"}, {"Key": TRANSCRIBE_OWNER_TAG, "Value": "user-a"}),
|
||||
}
|
||||
assert body == {"TranscriptionJobName": "j", "Tags": [{"Key": "env", "Value": "qa"}]}
|
||||
|
||||
def test_a_request_without_tags_gets_the_owner_tag(self):
|
||||
owned = transcribe_owned_start_request({"TranscriptionJobName": "j"}, VIRTUAL_KEY)
|
||||
|
||||
assert owned == {"TranscriptionJobName": "j", "Tags": ({"Key": TRANSCRIBE_OWNER_TAG, "Value": "user-a"},)}
|
||||
|
||||
def test_the_caller_cannot_supply_the_owner_tag(self):
|
||||
owned = transcribe_owned_start_request(
|
||||
{"TranscriptionJobName": "j", "Tags": [{"Key": TRANSCRIBE_OWNER_TAG, "Value": "user-b"}]}, VIRTUAL_KEY
|
||||
)
|
||||
|
||||
assert isinstance(owned, TranscribeRefusal)
|
||||
assert owned.status_code == 400
|
||||
|
||||
@pytest.mark.parametrize("tags", ["env=qa", ["env"], {"Key": "env"}])
|
||||
def test_malformed_tags_are_refused(self, tags: object):
|
||||
owned = transcribe_owned_start_request({"TranscriptionJobName": "j", "Tags": tags}, VIRTUAL_KEY)
|
||||
|
||||
assert isinstance(owned, TranscribeRefusal)
|
||||
assert owned.status_code == 400
|
||||
|
||||
def test_a_key_without_any_identity_is_refused(self):
|
||||
owned = transcribe_owned_start_request({"TranscriptionJobName": "j"}, UserAPIKeyAuth())
|
||||
|
||||
assert isinstance(owned, TranscribeRefusal)
|
||||
assert owned.status_code == 400
|
||||
|
||||
|
||||
def _tagged(owner: str | None) -> dict[str, object]:
|
||||
tags = {"Tags": [{"Key": TRANSCRIBE_OWNER_TAG, "Value": owner}]} if owner is not None else {}
|
||||
return _job("COMPLETED", **tags)
|
||||
|
||||
|
||||
class TestTranscribeJobAccessRefusal:
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_key_that_started_the_job_may_read_it(self):
|
||||
get_job, seen = _sequence(_tagged("user-a"))
|
||||
|
||||
assert await transcribe_job_access_refusal("job-1", VIRTUAL_KEY, get_job) is None
|
||||
assert seen == ["job-1"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_job_started_by_another_key_is_reported_missing(self):
|
||||
get_job, _ = _sequence(_tagged("user-b"))
|
||||
|
||||
refusal = await transcribe_job_access_refusal("job-1", VIRTUAL_KEY, get_job)
|
||||
|
||||
assert refusal is not None
|
||||
assert refusal.status_code == 404
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_job_started_outside_the_proxy_is_reported_missing(self):
|
||||
get_job, _ = _sequence(_tagged(None))
|
||||
|
||||
refusal = await transcribe_job_access_refusal("job-1", OTHER_VIRTUAL_KEY, get_job)
|
||||
|
||||
assert refusal is not None
|
||||
assert refusal.status_code == 404
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_job_that_cannot_be_looked_up_is_reported_missing(self):
|
||||
async def get_job(job_name: str) -> dict[str, object]:
|
||||
raise httpx.HTTPStatusError("boom", request=MagicMock(), response=MagicMock())
|
||||
|
||||
refusal = await transcribe_job_access_refusal("job-1", VIRTUAL_KEY, get_job)
|
||||
|
||||
assert refusal is not None
|
||||
assert refusal.status_code == 404
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_non_string_job_name_is_refused_before_any_lookup(self):
|
||||
get_job, seen = _sequence(_tagged("user-a"))
|
||||
|
||||
refusal = await transcribe_job_access_refusal(["job-1"], VIRTUAL_KEY, get_job)
|
||||
|
||||
assert refusal is not None
|
||||
assert refusal.status_code == 400
|
||||
assert seen == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_proxy_admin_reads_any_job_without_a_lookup(self):
|
||||
get_job, seen = _sequence(_tagged("user-b"))
|
||||
|
||||
assert await transcribe_job_access_refusal("job-1", ADMIN_KEY, get_job) is None
|
||||
assert seen == []
|
||||
|
||||
|
||||
class TestTranscribePassthroughHandler:
|
||||
def test_records_model_provider_and_the_given_cost(self):
|
||||
logging_obj = _make_logging_obj()
|
||||
request_body = {"TranscriptionJobName": "litellm-job-1"}
|
||||
|
||||
handler_result = TranscribePassthroughLoggingHandler.transcribe_passthrough_handler(
|
||||
httpx_response=_make_response("StartTranscriptionJob"),
|
||||
logging_obj=logging_obj,
|
||||
url_route="https://transcribe.us-west-2.amazonaws.com/",
|
||||
result='{"TranscriptionJob": {}}',
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
cache_hit=False,
|
||||
request_body=request_body,
|
||||
response_cost=0.0018,
|
||||
)
|
||||
|
||||
assert handler_result["result"] == {"response": '{"TranscriptionJob": {}}'}
|
||||
assert handler_result["kwargs"]["model"] == "transcribe/StartTranscriptionJob"
|
||||
assert handler_result["kwargs"]["custom_llm_provider"] == "transcribe"
|
||||
assert handler_result["kwargs"]["response_cost"] == 0.0018
|
||||
assert handler_result["kwargs"]["standard_logging_object"]["response_cost"] == 0.0018
|
||||
assert logging_obj.model_call_details["model"] == "transcribe/StartTranscriptionJob"
|
||||
assert logging_obj.model_call_details["custom_llm_provider"] == "transcribe"
|
||||
assert logging_obj.model_call_details["response_cost"] == 0.0018
|
||||
assert request_body == {"TranscriptionJobName": "litellm-job-1"}
|
||||
|
||||
def test_read_only_operations_default_to_zero_cost(self):
|
||||
handler_result = TranscribePassthroughLoggingHandler.transcribe_passthrough_handler(
|
||||
httpx_response=_make_response("GetTranscriptionJob"),
|
||||
logging_obj=_make_logging_obj(),
|
||||
url_route="https://transcribe.us-west-2.amazonaws.com/",
|
||||
result='{"TranscriptionJob": {}}',
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
cache_hit=False,
|
||||
request_body={"TranscriptionJobName": "litellm-job-1"},
|
||||
)
|
||||
|
||||
assert handler_result["kwargs"]["response_cost"] == 0.0
|
||||
|
||||
|
||||
class TestStartTranscriptionJobIsLoggedAtJobCost:
|
||||
@pytest.mark.asyncio
|
||||
async def test_success_handler_defers_logging_until_the_job_is_priced(self):
|
||||
priced: list[tuple[str, str, float, TranscriptionJobRecord | None]] = []
|
||||
|
||||
async def job_pricer(
|
||||
job_name: str, aws_region_name: str, cost_per_second: float, started_job: TranscriptionJobRecord | None
|
||||
) -> float:
|
||||
priced.append((job_name, aws_region_name, cost_per_second, started_job))
|
||||
return 0.0018
|
||||
|
||||
logged: list[dict[str, object]] = []
|
||||
|
||||
async def log(**kwargs: object) -> None:
|
||||
logged.append(kwargs)
|
||||
|
||||
handler = TranscribePassthroughLoggingHandler(job_pricer=job_pricer)
|
||||
logging_obj = _make_logging_obj()
|
||||
task = handler.schedule_priced_job_logging(
|
||||
httpx_response=_make_response("StartTranscriptionJob"),
|
||||
response_body={"TranscriptionJob": {}},
|
||||
logging_obj=logging_obj,
|
||||
url_route="https://transcribe.us-west-2.amazonaws.com/",
|
||||
result='{"TranscriptionJob": {}}',
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
cache_hit=False,
|
||||
request_body={"TranscriptionJobName": "litellm-job-1"},
|
||||
log=log,
|
||||
standard_pass_through_logging_payload={"cost_per_request": None},
|
||||
)
|
||||
await task
|
||||
|
||||
assert priced == [("litellm-job-1", "us-west-2", transcribe_cost_per_second(), TranscriptionJobRecord())]
|
||||
assert len(logged) == 1
|
||||
assert logged[0]["response_cost"] == 0.0018
|
||||
assert logged[0]["model"] == "transcribe/StartTranscriptionJob"
|
||||
assert logged[0]["standard_pass_through_logging_payload"] == {"cost_per_request": None}
|
||||
assert logging_obj.model_call_details["response_cost"] == 0.0018
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_job_is_not_logged_for_free_when_the_rate_leaves_the_cost_map(self, monkeypatch: pytest.MonkeyPatch):
|
||||
async def job_pricer(
|
||||
job_name: str, aws_region_name: str, cost_per_second: float, started_job: TranscriptionJobRecord | None
|
||||
) -> float:
|
||||
raise AssertionError("pricer must not run without a rate")
|
||||
|
||||
logged: list[dict[str, object]] = []
|
||||
|
||||
async def log(**kwargs: object) -> None:
|
||||
logged.append(kwargs)
|
||||
|
||||
monkeypatch.delitem(litellm.model_cost, "transcribe/StartTranscriptionJob")
|
||||
await TranscribePassthroughLoggingHandler(job_pricer=job_pricer).schedule_priced_job_logging(
|
||||
httpx_response=_make_response("StartTranscriptionJob"),
|
||||
response_body={"TranscriptionJob": {}},
|
||||
logging_obj=_make_logging_obj(),
|
||||
url_route="https://transcribe.us-west-2.amazonaws.com/",
|
||||
result='{"TranscriptionJob": {}}',
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
cache_hit=False,
|
||||
request_body={"TranscriptionJobName": "litellm-job-1"},
|
||||
log=log,
|
||||
)
|
||||
|
||||
assert logged == []
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pass_through_success_handler_routes_job_starts_to_the_pricer(self):
|
||||
scheduled: list[str] = []
|
||||
|
||||
async def job_pricer(
|
||||
job_name: str, aws_region_name: str, cost_per_second: float, started_job: TranscriptionJobRecord | None
|
||||
) -> float:
|
||||
scheduled.append(job_name)
|
||||
return 0.0
|
||||
|
||||
immediate: list[dict[str, object]] = []
|
||||
|
||||
async def log_dispatch(**kwargs: object) -> None:
|
||||
immediate.append(kwargs)
|
||||
|
||||
logging = PassThroughEndpointLogging(
|
||||
TranscribePassthroughLoggingHandler(job_pricer=job_pricer), log_dispatch=log_dispatch
|
||||
)
|
||||
|
||||
await logging.pass_through_async_success_handler(
|
||||
httpx_response=_make_response("StartTranscriptionJob"),
|
||||
response_body={"TranscriptionJob": {}},
|
||||
logging_obj=_make_logging_obj(),
|
||||
url_route="https://transcribe.us-west-2.amazonaws.com/",
|
||||
result='{"TranscriptionJob": {}}',
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
cache_hit=False,
|
||||
request_body={"TranscriptionJobName": "litellm-job-1"},
|
||||
passthrough_logging_payload={"url": "https://transcribe.us-west-2.amazonaws.com/"},
|
||||
custom_llm_provider="transcribe",
|
||||
)
|
||||
await asyncio.gather(*logging.transcribe_passthrough_logging_handler._pricing_tasks)
|
||||
|
||||
assert scheduled == ["litellm-job-1"]
|
||||
assert [entry["response_cost"] for entry in immediate] == [0.0]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pass_through_success_handler_prices_a_relayed_start_response_from_its_parsed_body(self):
|
||||
started_jobs: list[TranscriptionJobRecord | None] = []
|
||||
logged_costs: list[object] = []
|
||||
|
||||
async def job_pricer(
|
||||
job_name: str, aws_region_name: str, cost_per_second: float, started_job: TranscriptionJobRecord | None
|
||||
) -> float:
|
||||
started_jobs.append(started_job)
|
||||
return 18 * COST_PER_SECOND
|
||||
|
||||
async def log_dispatch(**kwargs: object) -> None:
|
||||
logged_costs.append(kwargs["response_cost"])
|
||||
|
||||
start_response = {
|
||||
"TranscriptionJob": {
|
||||
"TranscriptionJobName": "litellm-job-1",
|
||||
"TranscriptionJobStatus": "IN_PROGRESS",
|
||||
"Media": {"MediaFileUri": "s3://b/started.wav"},
|
||||
"CreationTime": 5.0,
|
||||
}
|
||||
}
|
||||
logging = PassThroughEndpointLogging(
|
||||
TranscribePassthroughLoggingHandler(job_pricer=job_pricer), log_dispatch=log_dispatch
|
||||
)
|
||||
|
||||
await logging.pass_through_async_success_handler(
|
||||
httpx_response=await _relayed_response("StartTranscriptionJob", json.dumps(start_response).encode()),
|
||||
response_body=start_response,
|
||||
logging_obj=_make_logging_obj(),
|
||||
url_route="https://transcribe.us-west-2.amazonaws.com/",
|
||||
result="",
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
cache_hit=False,
|
||||
request_body={"TranscriptionJobName": "litellm-job-1"},
|
||||
passthrough_logging_payload={"url": "https://transcribe.us-west-2.amazonaws.com/"},
|
||||
custom_llm_provider="transcribe",
|
||||
)
|
||||
await asyncio.gather(*logging.transcribe_passthrough_logging_handler._pricing_tasks)
|
||||
|
||||
assert started_jobs == [
|
||||
TranscriptionJobRecord(
|
||||
TranscriptionJobStatus="IN_PROGRESS", CreationTime=5.0, Media={"MediaFileUri": "s3://b/started.wav"}
|
||||
)
|
||||
]
|
||||
assert logged_costs == [pytest.approx(18 * COST_PER_SECOND)]
|
||||
|
||||
|
||||
class TestIsTranscribeRoute:
|
||||
def test_matches_by_provider_tag(self):
|
||||
assert PassThroughEndpointLogging().is_transcribe_route("transcribe")
|
||||
|
||||
def test_does_not_match_other_providers(self):
|
||||
assert not PassThroughEndpointLogging().is_transcribe_route("comprehendmedical")
|
||||
|
||||
def test_dispatch_reaches_transcribe_handler(self):
|
||||
logging_obj = _make_logging_obj()
|
||||
|
||||
normalized = PassThroughEndpointLogging().normalize_llm_passthrough_logging_payload(
|
||||
httpx_response=_make_response("GetTranscriptionJob"),
|
||||
response_body={"TranscriptionJob": {}},
|
||||
request_body={"TranscriptionJobName": "litellm-job-1"},
|
||||
logging_obj=logging_obj,
|
||||
url_route="https://transcribe.us-west-2.amazonaws.com/",
|
||||
result='{"TranscriptionJob": {}}',
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
cache_hit=False,
|
||||
custom_llm_provider="transcribe",
|
||||
)
|
||||
|
||||
assert normalized["kwargs"]["model"] == "transcribe/GetTranscriptionJob"
|
||||
assert normalized["kwargs"]["response_cost"] == 0.0
|
||||
|
||||
def test_config_driven_passthrough_to_transcribe_host_is_not_claimed(self):
|
||||
logging_obj = _make_logging_obj()
|
||||
|
||||
normalized = PassThroughEndpointLogging().normalize_llm_passthrough_logging_payload(
|
||||
httpx_response=_make_response("GetTranscriptionJob"),
|
||||
response_body={"TranscriptionJob": {}},
|
||||
request_body={"TranscriptionJobName": "litellm-job-1"},
|
||||
logging_obj=logging_obj,
|
||||
url_route="https://transcribe.us-west-2.amazonaws.com/",
|
||||
result='{"TranscriptionJob": {}}',
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
cache_hit=False,
|
||||
custom_llm_provider=None,
|
||||
)
|
||||
|
||||
assert normalized["kwargs"].get("model") != "transcribe/GetTranscriptionJob"
|
||||
|
|
@ -29,6 +29,7 @@ from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
|
|||
BaseOpenAIPassThroughHandler,
|
||||
RouteChecks,
|
||||
_join_url_paths,
|
||||
_proxy_general_settings,
|
||||
anthropic_proxy_route,
|
||||
azure_proxy_route,
|
||||
bedrock_llm_proxy_route,
|
||||
|
|
@ -5275,6 +5276,304 @@ class TestComprehendMedicalProxyRoute:
|
|||
assert exc_info.value.status_code == 400
|
||||
|
||||
|
||||
TRANSCRIBE_UPSTREAM = "https://transcribe.us-west-2.amazonaws.com/"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def transcribe_client(monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
|
||||
from litellm.proxy.proxy_server import app
|
||||
|
||||
monkeypatch.setenv("AWS_REGION_NAME", "us-west-2")
|
||||
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "test-access-key")
|
||||
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "test-secret-key")
|
||||
monkeypatch.delenv("AWS_SESSION_TOKEN", raising=False)
|
||||
monkeypatch.delenv("SERVER_ROOT_PATH", raising=False)
|
||||
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
|
||||
litellm.in_memory_llm_clients_cache.flush_cache()
|
||||
monkeypatch.setitem(
|
||||
app.dependency_overrides, user_api_key_auth, lambda: UserAPIKeyAuth(api_key="sk-virtual", user_id="user-a")
|
||||
)
|
||||
monkeypatch.setitem(
|
||||
app.dependency_overrides, _proxy_general_settings, lambda: {"transcribe_media_buckets": ["bucket"]}
|
||||
)
|
||||
yield TestClient(app)
|
||||
|
||||
|
||||
def _owned_job(owner: str | None, status: str = "COMPLETED") -> dict[str, object]:
|
||||
tags = {"Tags": [{"Key": "litellm-owner", "Value": owner}]} if owner is not None else {}
|
||||
return {"TranscriptionJob": {"TranscriptionJobName": "litellm-job-1", "TranscriptionJobStatus": status, **tags}}
|
||||
|
||||
|
||||
class TestTranscribeProxyRoute:
|
||||
START_JOB_BODY: Final = MappingProxyType(
|
||||
{
|
||||
"TranscriptionJobName": "litellm-job-1",
|
||||
"LanguageCode": "en-US",
|
||||
"Media": {"MediaFileUri": "s3://bucket/audio.wav"},
|
||||
}
|
||||
)
|
||||
OWNER_TAG: Final = MappingProxyType({"Key": "litellm-owner", "Value": "user-a"})
|
||||
|
||||
def test_signs_and_forwards_start_transcription_job(self, transcribe_client: TestClient) -> None:
|
||||
upstream_body = {
|
||||
"TranscriptionJob": {"TranscriptionJobName": "litellm-job-1", "TranscriptionJobStatus": "IN_PROGRESS"}
|
||||
}
|
||||
with respx.mock(assert_all_called=True) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM).mock(return_value=httpx.Response(200, json=upstream_body))
|
||||
response = transcribe_client.post(
|
||||
"/transcribe/StartTranscriptionJob",
|
||||
json=dict(self.START_JOB_BODY),
|
||||
headers={"Authorization": "Bearer sk-virtual"},
|
||||
)
|
||||
|
||||
assert (response.status_code, response.json()) == (200, upstream_body)
|
||||
targets = [call.request.headers["x-amz-target"] for call in route.calls]
|
||||
assert targets[0] == "Transcribe.StartTranscriptionJob"
|
||||
assert set(targets[1:]) <= {"Transcribe.GetTranscriptionJob"}
|
||||
sent = route.calls[0].request
|
||||
assert json.loads(sent.content) == {**dict(self.START_JOB_BODY), "Tags": [dict(self.OWNER_TAG)]}
|
||||
assert sent.headers["content-type"] == "application/x-amz-json-1.1"
|
||||
assert sent.headers["authorization"].startswith("AWS4-HMAC-SHA256 Credential=test-access-key/")
|
||||
assert "/us-west-2/transcribe/aws4_request" in sent.headers["authorization"]
|
||||
assert "x-amz-date" in sent.headers
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"body, member",
|
||||
[
|
||||
({"Media": {"MediaFileUri": "s3://other-tenant/audio.wav"}}, "Media.MediaFileUri"),
|
||||
({"OutputBucketName": "other-tenant"}, "OutputBucketName"),
|
||||
({"DataAccessRoleArn": "arn:aws:iam::123456789012:role/reader"}, "DataAccessRoleArn"),
|
||||
],
|
||||
)
|
||||
def test_storage_outside_the_listed_buckets_is_refused_before_signing(
|
||||
self, transcribe_client: TestClient, body: dict[str, object], member: str
|
||||
) -> None:
|
||||
with respx.mock(assert_all_called=False) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM)
|
||||
response = transcribe_client.post("/transcribe/StartTranscriptionJob", json={**dict(self.START_JOB_BODY), **body})
|
||||
|
||||
assert response.status_code == 403
|
||||
assert member in response.json()["detail"]
|
||||
assert not route.called
|
||||
|
||||
def test_start_needs_a_bucket_list_unless_the_caller_is_a_proxy_admin(
|
||||
self, transcribe_client: TestClient, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
from litellm.proxy.proxy_server import app
|
||||
|
||||
monkeypatch.setitem(app.dependency_overrides, _proxy_general_settings, lambda: {})
|
||||
with respx.mock(assert_all_called=False) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM).mock(return_value=httpx.Response(200, json=_owned_job("admin")))
|
||||
refused = transcribe_client.post("/transcribe/StartTranscriptionJob", json=dict(self.START_JOB_BODY))
|
||||
monkeypatch.setitem(
|
||||
app.dependency_overrides,
|
||||
user_api_key_auth,
|
||||
lambda: UserAPIKeyAuth(api_key="sk-admin", user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN),
|
||||
)
|
||||
allowed = transcribe_client.post("/transcribe/StartTranscriptionJob", json=dict(self.START_JOB_BODY))
|
||||
|
||||
assert refused.status_code == 403
|
||||
assert "transcribe_media_buckets" in refused.json()["detail"]
|
||||
assert allowed.status_code == 200
|
||||
assert route.calls[0].request.headers["x-amz-target"] == "Transcribe.StartTranscriptionJob"
|
||||
|
||||
def test_the_caller_cannot_forge_the_owner_tag(self, transcribe_client: TestClient) -> None:
|
||||
with respx.mock(assert_all_called=False) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM)
|
||||
response = transcribe_client.post(
|
||||
"/transcribe/StartTranscriptionJob",
|
||||
json={**dict(self.START_JOB_BODY), "Tags": [{"Key": "litellm-owner", "Value": "user-b"}]},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "litellm-owner" in response.json()["detail"]
|
||||
assert not route.called
|
||||
|
||||
def test_sdk_route_reads_operation_from_x_amz_target_and_resigns(self, transcribe_client: TestClient) -> None:
|
||||
with respx.mock(assert_all_called=True) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM).mock(return_value=httpx.Response(200, json=_owned_job("user-a")))
|
||||
response = transcribe_client.post(
|
||||
"/transcribe",
|
||||
json={"TranscriptionJobName": "litellm-job-1"},
|
||||
headers={
|
||||
"Authorization": "AWS4-HMAC-SHA256 Credential=sk-virtual/20260101/us-west-2/transcribe/aws4_request",
|
||||
"X-Amz-Target": "Transcribe.GetTranscriptionJob",
|
||||
"Content-Type": "application/x-amz-json-1.1",
|
||||
},
|
||||
)
|
||||
|
||||
assert (response.status_code, response.json()) == (200, _owned_job("user-a"))
|
||||
assert [call.request.headers["x-amz-target"] for call in route.calls] == ["Transcribe.GetTranscriptionJob"] * 2
|
||||
sent = route.calls.last.request
|
||||
assert "Credential=test-access-key/" in sent.headers["authorization"]
|
||||
assert "sk-virtual" not in sent.headers["authorization"]
|
||||
|
||||
@pytest.mark.parametrize("operation", ["GetTranscriptionJob", "DeleteTranscriptionJob"])
|
||||
@pytest.mark.parametrize("owner", ["user-b", None])
|
||||
def test_jobs_started_by_others_are_not_reachable(
|
||||
self, transcribe_client: TestClient, operation: str, owner: str | None
|
||||
) -> None:
|
||||
with respx.mock(assert_all_called=True) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM).mock(return_value=httpx.Response(200, json=_owned_job(owner)))
|
||||
response = transcribe_client.post(
|
||||
f"/transcribe/{operation}", json={"TranscriptionJobName": "litellm-job-1"}
|
||||
)
|
||||
|
||||
assert response.status_code == 404
|
||||
assert [call.request.headers["x-amz-target"] for call in route.calls] == ["Transcribe.GetTranscriptionJob"]
|
||||
|
||||
def test_the_owner_may_delete_the_job(self, transcribe_client: TestClient) -> None:
|
||||
with respx.mock(assert_all_called=True) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM)
|
||||
route.side_effect = [httpx.Response(200, json=_owned_job("user-a")), httpx.Response(200, json={})]
|
||||
response = transcribe_client.post(
|
||||
"/transcribe/DeleteTranscriptionJob", json={"TranscriptionJobName": "litellm-job-1"}
|
||||
)
|
||||
|
||||
assert (response.status_code, response.json()) == (200, {})
|
||||
assert [call.request.headers["x-amz-target"] for call in route.calls] == [
|
||||
"Transcribe.GetTranscriptionJob",
|
||||
"Transcribe.DeleteTranscriptionJob",
|
||||
]
|
||||
|
||||
@pytest.mark.parametrize("operation", ["ListTranscriptionJobs", "ListVocabularies", "DeleteVocabulary"])
|
||||
def test_account_wide_operations_need_a_proxy_admin(self, transcribe_client: TestClient, operation: str) -> None:
|
||||
with respx.mock(assert_all_called=False) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM)
|
||||
response = transcribe_client.post(f"/transcribe/{operation}", json={})
|
||||
|
||||
assert response.status_code == 403
|
||||
assert operation in response.json()["detail"]
|
||||
assert not route.called
|
||||
|
||||
def test_a_proxy_admin_reaches_account_wide_operations(
|
||||
self, transcribe_client: TestClient, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
from litellm.proxy._types import LitellmUserRoles
|
||||
from litellm.proxy.proxy_server import app
|
||||
|
||||
monkeypatch.setitem(
|
||||
app.dependency_overrides,
|
||||
user_api_key_auth,
|
||||
lambda: UserAPIKeyAuth(api_key="sk-admin", user_role=LitellmUserRoles.PROXY_ADMIN),
|
||||
)
|
||||
with respx.mock(assert_all_called=True) as upstream:
|
||||
upstream.post(TRANSCRIBE_UPSTREAM).mock(
|
||||
return_value=httpx.Response(200, json={"TranscriptionJobSummaries": []})
|
||||
)
|
||||
response = transcribe_client.post("/transcribe/ListTranscriptionJobs", json={})
|
||||
|
||||
assert (response.status_code, response.json()) == (200, {"TranscriptionJobSummaries": []})
|
||||
|
||||
def test_upstream_error_status_and_body_are_returned(self, transcribe_client: TestClient) -> None:
|
||||
aws_error = {"__type": "BadRequestException", "Message": "The requested job couldn't be found."}
|
||||
with respx.mock(assert_all_called=True) as upstream:
|
||||
upstream.post(TRANSCRIBE_UPSTREAM).mock(return_value=httpx.Response(400, json=aws_error))
|
||||
response = transcribe_client.post(
|
||||
"/transcribe/StartTranscriptionJob",
|
||||
json={**dict(self.START_JOB_BODY), "TranscriptionJobName": "missing"},
|
||||
)
|
||||
|
||||
assert (response.status_code, response.json()) == (400, aws_error)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"operation",
|
||||
[
|
||||
"Start-Transcription-Job",
|
||||
"Transcribe.StartTranscriptionJob",
|
||||
"a" * 200,
|
||||
"starttranscriptionjob",
|
||||
"DetectEntitiesV2",
|
||||
],
|
||||
)
|
||||
def test_rejects_unsupported_operations_without_calling_aws(
|
||||
self, transcribe_client: TestClient, operation: str
|
||||
) -> None:
|
||||
with respx.mock(assert_all_called=False) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM)
|
||||
response = transcribe_client.post(f"/transcribe/{operation}", json={})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "Unsupported Amazon Transcribe operation" in response.json()["detail"]
|
||||
assert not route.called
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw_body",
|
||||
['{"MaxResults": 5, "stream": true}', '{"MaxResults": 5, "stream": false}', '["x"]', "not json"],
|
||||
)
|
||||
def test_rejects_bad_bodies_without_calling_aws(self, transcribe_client: TestClient, raw_body: str) -> None:
|
||||
with respx.mock(assert_all_called=False) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM)
|
||||
response = transcribe_client.post(
|
||||
"/transcribe/GetTranscriptionJob", content=raw_body, headers={"Content-Type": "application/json"}
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
assert not route.called
|
||||
|
||||
def test_missing_region_returns_400_without_calling_aws(
|
||||
self, transcribe_client: TestClient, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
for name in ("AWS_REGION_NAME", "AWS_REGION", "AWS_DEFAULT_REGION"):
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
with respx.mock(assert_all_called=False) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM)
|
||||
response = transcribe_client.post("/transcribe/GetTranscriptionJob", json={})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "AWS region" in response.json()["detail"]
|
||||
assert not route.called
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("operation", "body", "detail_fragment"),
|
||||
[
|
||||
("StartMedicalTranscriptionJob", {"MedicalTranscriptionJobName": "j"}, "StartMedicalTranscriptionJob"),
|
||||
("StartCallAnalyticsJob", {"CallAnalyticsJobName": "j"}, "StartCallAnalyticsJob"),
|
||||
("StartMedicalScribeJob", {"MedicalScribeJobName": "j"}, "StartMedicalScribeJob"),
|
||||
("StartTranscriptionJob", {"ContentRedaction": {"RedactionType": "PII"}}, "ContentRedaction"),
|
||||
("StartTranscriptionJob", {"ToxicityDetection": [{"ToxicityCategories": ["ALL"]}]}, "ToxicityDetection"),
|
||||
("StartTranscriptionJob", {"ModelSettings": {"LanguageModelName": "clm"}}, "LanguageModelName"),
|
||||
],
|
||||
)
|
||||
def test_rejects_unpriced_billable_jobs_without_calling_aws(
|
||||
self, transcribe_client: TestClient, operation: str, body: dict[str, object], detail_fragment: str
|
||||
) -> None:
|
||||
with respx.mock(assert_all_called=False) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM)
|
||||
response = transcribe_client.post(f"/transcribe/{operation}", json={**dict(self.START_JOB_BODY), **body})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert detail_fragment in response.json()["detail"]
|
||||
assert not route.called
|
||||
|
||||
def test_rejects_start_transcription_job_when_the_cost_map_has_no_rate(
|
||||
self, transcribe_client: TestClient, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.delitem(litellm.model_cost, "transcribe/StartTranscriptionJob")
|
||||
with respx.mock(assert_all_called=False) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM)
|
||||
response = transcribe_client.post("/transcribe/StartTranscriptionJob", json=dict(self.START_JOB_BODY))
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "model cost map" in response.json()["detail"]
|
||||
assert not route.called
|
||||
|
||||
@pytest.mark.parametrize("target_header", ["", "Transcribe", "ComprehendMedical_20181030.DetectPHI", "Transcribe."])
|
||||
def test_sdk_route_rejects_bad_x_amz_target(self, transcribe_client: TestClient, target_header: str) -> None:
|
||||
with respx.mock(assert_all_called=False) as upstream:
|
||||
route = upstream.post(TRANSCRIBE_UPSTREAM)
|
||||
response = transcribe_client.post("/transcribe", json={}, headers={"X-Amz-Target": target_header})
|
||||
|
||||
assert response.status_code == 400
|
||||
assert "X-Amz-Target" in response.json()["detail"]
|
||||
assert not route.called
|
||||
|
||||
def test_transcribe_is_a_mapped_pass_through_route(self) -> None:
|
||||
from litellm.proxy._types import LiteLLMRoutes
|
||||
|
||||
assert "/transcribe" in LiteLLMRoutes.mapped_pass_through_routes.value
|
||||
|
||||
|
||||
LIVE_RESOURCE_PATH = "projects/proj-db/locations/global/publishers/google/models/gemini-live-2.5-flash"
|
||||
|
||||
|
||||
|
|
@ -5315,9 +5614,7 @@ class TestVertexAILiveWebsocketPassthrough:
|
|||
]
|
||||
)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", llm_router)
|
||||
monkeypatch.setattr(
|
||||
passthrough_module.passthrough_endpoint_router, "default_vertex_config", None
|
||||
)
|
||||
monkeypatch.setattr(passthrough_module.passthrough_endpoint_router, "default_vertex_config", None)
|
||||
self._clear_vertex_env(monkeypatch)
|
||||
websocket = self._websocket()
|
||||
ensure_token = AsyncMock(return_value=("token-abc", "proj-db"))
|
||||
|
|
@ -5459,9 +5756,7 @@ class TestVertexAILiveWebsocketPassthrough:
|
|||
)
|
||||
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.llm_router", None)
|
||||
monkeypatch.setattr(
|
||||
passthrough_module.passthrough_endpoint_router, "default_vertex_config", None
|
||||
)
|
||||
monkeypatch.setattr(passthrough_module.passthrough_endpoint_router, "default_vertex_config", None)
|
||||
self._clear_vertex_env(monkeypatch)
|
||||
websocket = self._websocket()
|
||||
ensure_token = AsyncMock(side_effect=Exception("Unable to find your credentials"))
|
||||
|
|
|
|||
|
|
@ -4407,12 +4407,14 @@ async def test_pass_through_request_relays_non_json_body_without_buffering():
|
|||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pass_through_request_json_response_stays_buffered_for_logging():
|
||||
@pytest.mark.parametrize("content_type", ["application/json", "application/x-amz-json-1.1"])
|
||||
async def test_pass_through_request_json_response_stays_buffered_for_logging(content_type: str):
|
||||
"""
|
||||
JSON responses (content-type application/json) must keep the buffered
|
||||
behavior: spend logging and guardrails inspect the parsed body, so the
|
||||
handler reads the full upstream body and passes the parsed dict to the
|
||||
success handler.
|
||||
JSON responses (content-type application/json, and the AWS JSON protocol
|
||||
media types AWS services such as Amazon Transcribe answer with) must keep
|
||||
the buffered behavior: spend logging and guardrails inspect the parsed body,
|
||||
so the handler reads the full upstream body and passes the parsed dict to
|
||||
the success handler instead of handing it a relayed, already closed response.
|
||||
"""
|
||||
from fastapi.responses import StreamingResponse
|
||||
|
||||
|
|
@ -4423,7 +4425,7 @@ async def test_pass_through_request_json_response_stays_buffered_for_logging():
|
|||
fake_client, cleanup = _inject_fake_passthrough_client(
|
||||
_FakeUpstreamTransport(
|
||||
status_code=200,
|
||||
headers={"content-type": "application/json"},
|
||||
headers={"content-type": content_type},
|
||||
stream=upstream_stream,
|
||||
),
|
||||
timeout=312.0,
|
||||
|
|
|
|||
|
|
@ -3746,6 +3746,27 @@ async def test_ProxyConfig__update_general_settings_yaml_allowed_file_extensions
|
|||
assert ps.general_settings.get("allowed_file_extensions") == [".pdf"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ProxyConfig__update_general_settings_applies_db_transcribe_media_buckets(monkeypatch):
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.general_settings", {})
|
||||
pc = ProxyConfig()
|
||||
await pc._update_general_settings({"transcribe_media_buckets": ["team-audio"]})
|
||||
from litellm.proxy import proxy_server as ps
|
||||
|
||||
assert ps.general_settings.get("transcribe_media_buckets") == ["team-audio"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ProxyConfig__update_general_settings_yaml_transcribe_media_buckets_wins_over_db(monkeypatch):
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.general_settings", {"transcribe_media_buckets": ["yaml-audio"]})
|
||||
pc = ProxyConfig()
|
||||
pc._yaml_general_settings_keys = {"transcribe_media_buckets"}
|
||||
await pc._update_general_settings({"transcribe_media_buckets": ["team-audio"]})
|
||||
from litellm.proxy import proxy_server as ps
|
||||
|
||||
assert ps.general_settings.get("transcribe_media_buckets") == ["yaml-audio"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ProxyConfig__update_general_settings_none_input_noop():
|
||||
pc = ProxyConfig()
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
import { renderWithProviders, screen, within } from "../../../../../tests/test-utils";
|
||||
import { fireEvent, renderWithProviders, screen, within } from "../../../../../tests/test-utils";
|
||||
import userEvent from "@testing-library/user-event";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import GeneralSettings from "./general_settings";
|
||||
|
|
@ -159,6 +159,56 @@ describe("GeneralSettings tabs", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("persists a List setting typed as comma-separated text as a trimmed string array", async () => {
|
||||
vi.mocked(getGeneralSettingsCall).mockResolvedValue([
|
||||
{
|
||||
field_name: "transcribe_media_buckets",
|
||||
field_type: "List",
|
||||
field_value: ["old-bucket"],
|
||||
field_description: "buckets",
|
||||
stored_in_db: true,
|
||||
},
|
||||
]);
|
||||
vi.mocked(updateConfigFieldSetting).mockClear();
|
||||
const user = userEvent.setup();
|
||||
renderWithProviders(<GeneralSettings accessToken="token" userRole="Admin" userID="user" />);
|
||||
await user.click(screen.getByRole("tab", { name: "General" }));
|
||||
const input = await screen.findByRole("textbox", { name: "transcribe_media_buckets" });
|
||||
expect(input).toHaveValue("old-bucket");
|
||||
fireEvent.change(input, { target: { value: " team-audio, shared.audio ,, " } });
|
||||
await user.click(
|
||||
within(screen.getByRole("row", { name: /transcribe_media_buckets/ })).getByRole("button", { name: "Update" }),
|
||||
);
|
||||
expect(vi.mocked(updateConfigFieldSetting).mock.calls).toEqual([
|
||||
["token", "transcribe_media_buckets", ["team-audio", "shared.audio"]],
|
||||
]);
|
||||
});
|
||||
|
||||
it("clears a stored List setting when Update is clicked on an emptied input", async () => {
|
||||
vi.mocked(getGeneralSettingsCall).mockResolvedValue([
|
||||
{
|
||||
field_name: "transcribe_media_buckets",
|
||||
field_type: "List",
|
||||
field_value: ["old-bucket"],
|
||||
field_description: "buckets",
|
||||
stored_in_db: true,
|
||||
},
|
||||
]);
|
||||
vi.mocked(updateConfigFieldSetting).mockClear();
|
||||
vi.mocked(deleteConfigFieldSetting).mockClear();
|
||||
const user = userEvent.setup();
|
||||
renderWithProviders(<GeneralSettings accessToken="token" userRole="Admin" userID="user" />);
|
||||
await user.click(screen.getByRole("tab", { name: "General" }));
|
||||
const input = await screen.findByRole("textbox", { name: "transcribe_media_buckets" });
|
||||
fireEvent.change(input, { target: { value: " , " } });
|
||||
await user.click(
|
||||
within(screen.getByRole("row", { name: /transcribe_media_buckets/ })).getByRole("button", { name: "Update" }),
|
||||
);
|
||||
expect(vi.mocked(deleteConfigFieldSetting).mock.calls).toEqual([["token", "transcribe_media_buckets"]]);
|
||||
expect(updateConfigFieldSetting).not.toHaveBeenCalled();
|
||||
expect(screen.getByRole("textbox", { name: "transcribe_media_buckets" })).toHaveValue("");
|
||||
});
|
||||
|
||||
it("should delete only the Default setting and retain explicit false and zero", async () => {
|
||||
vi.mocked(getGeneralSettingsCall).mockResolvedValue([
|
||||
{
|
||||
|
|
|
|||
|
|
@ -43,6 +43,16 @@ const NUMERIC_INPUT_WIDTH = "w-36";
|
|||
|
||||
const toNumericValue = (raw: string): number | null => (raw === "" ? null : Number(raw));
|
||||
|
||||
const toListValue = (raw: string): string[] | null => {
|
||||
const items = raw
|
||||
.split(",")
|
||||
.map((item) => item.trim())
|
||||
.filter((item) => item !== "");
|
||||
return items.length === 0 ? null : items;
|
||||
};
|
||||
|
||||
const fromListValue = (value: unknown): string => (Array.isArray(value) ? value.join(", ") : "");
|
||||
|
||||
const SettingValueEditor: React.FC<{
|
||||
setting: generalSettingsItem;
|
||||
onChange: (fieldName: string, newValue: any) => void;
|
||||
|
|
@ -93,6 +103,17 @@ const SettingValueEditor: React.FC<{
|
|||
</InputGroup>
|
||||
);
|
||||
}
|
||||
if (setting.field_type === "List") {
|
||||
return (
|
||||
<Input
|
||||
key={String(setting.stored_in_db)}
|
||||
aria-label={setting.field_name}
|
||||
placeholder="Comma-separated values"
|
||||
defaultValue={fromListValue(setting.field_value)}
|
||||
onChange={(event) => onChange(setting.field_name, toListValue(event.target.value))}
|
||||
/>
|
||||
);
|
||||
}
|
||||
if (setting.field_type === "Select") {
|
||||
return (
|
||||
<Select value={setting.field_value ?? null} onValueChange={(newValue) => onChange(setting.field_name, newValue)}>
|
||||
|
|
@ -228,7 +249,7 @@ const GeneralSettings: React.FC<GeneralSettingsPageProps> = ({ accessToken, user
|
|||
const fieldValue = setting?.field_value;
|
||||
|
||||
if (fieldValue == null) {
|
||||
if (setting?.field_type === "Select") handleResetField(fieldName);
|
||||
if (setting?.field_type === "Select" || setting?.field_type === "List") handleResetField(fieldName);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
|
|
|
|||
111
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
111
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
|
|
@ -16476,6 +16476,61 @@ export interface paths {
|
|||
patch: operations["toolset_mcp_route_toolset__toolset_name__mcp_patch"];
|
||||
trace?: never;
|
||||
};
|
||||
"/transcribe": {
|
||||
parameters: {
|
||||
query?: never;
|
||||
header?: never;
|
||||
path?: never;
|
||||
cookie?: never;
|
||||
};
|
||||
get?: never;
|
||||
put?: never;
|
||||
/**
|
||||
* Transcribe Sdk Proxy Route
|
||||
* @description AWS-SDK-shaped pass-through for Amazon Transcribe: point the SDK's `endpoint_url`
|
||||
* at `/transcribe` and the operation is read from the `X-Amz-Target` header, per the
|
||||
* AWS JSON 1.1 protocol.
|
||||
*
|
||||
* [Docs](https://docs.litellm.ai/docs/pass_through/transcribe)
|
||||
*/
|
||||
post: operations["transcribe_sdk_proxy_route_transcribe_post"];
|
||||
delete?: never;
|
||||
options?: never;
|
||||
head?: never;
|
||||
patch?: never;
|
||||
trace?: never;
|
||||
};
|
||||
"/transcribe/{operation}": {
|
||||
parameters: {
|
||||
query?: never;
|
||||
header?: never;
|
||||
path?: never;
|
||||
cookie?: never;
|
||||
};
|
||||
get?: never;
|
||||
put?: never;
|
||||
/**
|
||||
* Transcribe Proxy Route
|
||||
* @description Pass-through for the Amazon Transcribe API, e.g. `POST /transcribe/StartTranscriptionJob`.
|
||||
*
|
||||
* The request body is forwarded to the AWS JSON 1.1 API and signed with SigV4 using the
|
||||
* proxy's AWS credentials. Standard jobs are tagged with the calling key's owner so that
|
||||
* only that owner (or a proxy admin) can read or delete them, and keys other than proxy
|
||||
* admins may only read media from and write transcripts to the S3 buckets listed in
|
||||
* `general_settings.transcribe_media_buckets`; account-wide operations
|
||||
* such as ListTranscriptionJobs are limited to proxy admins. Streaming transcription
|
||||
* (`transcribestreaming`) uses a separate HTTP/2 event-stream protocol and is not served
|
||||
* by this route.
|
||||
*
|
||||
* [Docs](https://docs.litellm.ai/docs/pass_through/transcribe)
|
||||
*/
|
||||
post: operations["transcribe_proxy_route_transcribe__operation__post"];
|
||||
delete?: never;
|
||||
options?: never;
|
||||
head?: never;
|
||||
patch?: never;
|
||||
trace?: never;
|
||||
};
|
||||
"/typesafe/{endpoint}": {
|
||||
parameters: {
|
||||
query?: never;
|
||||
|
|
@ -26829,6 +26884,11 @@ export interface components {
|
|||
* @description Fine-grained control over which object types to load from the database when store_model_in_db is True. Available types: 'models', 'mcp', 'guardrails', 'vector_stores', 'pass_through_endpoints', 'prompts', 'model_cost_map', 'tools', 'config_overrides'. If not set, all objects are loaded (default behavior).
|
||||
*/
|
||||
supported_db_objects?: components["schemas"]["SupportedDBObjectType"][] | null;
|
||||
/**
|
||||
* Transcribe Media Buckets
|
||||
* @description S3 bucket names that keys other than proxy admins may read media from and write transcripts to through the Amazon Transcribe pass-through. Unset means only proxy admins can start transcription jobs.
|
||||
*/
|
||||
transcribe_media_buckets?: string[] | null;
|
||||
/**
|
||||
* Trusted Proxy Ranges
|
||||
* @description CIDR ranges of trusted reverse proxies allowed to provide identity headers for header-based auth paths such as enable_oauth2_proxy_auth and custom_ui_sso_sign_in_handler.
|
||||
|
|
@ -61698,6 +61758,57 @@ export interface operations {
|
|||
};
|
||||
};
|
||||
};
|
||||
transcribe_sdk_proxy_route_transcribe_post: {
|
||||
parameters: {
|
||||
query?: never;
|
||||
header?: never;
|
||||
path?: never;
|
||||
cookie?: never;
|
||||
};
|
||||
requestBody?: never;
|
||||
responses: {
|
||||
/** @description Successful Response */
|
||||
200: {
|
||||
headers: {
|
||||
[name: string]: unknown;
|
||||
};
|
||||
content: {
|
||||
"application/json": unknown;
|
||||
};
|
||||
};
|
||||
};
|
||||
};
|
||||
transcribe_proxy_route_transcribe__operation__post: {
|
||||
parameters: {
|
||||
query?: never;
|
||||
header?: never;
|
||||
path: {
|
||||
operation: string;
|
||||
};
|
||||
cookie?: never;
|
||||
};
|
||||
requestBody?: never;
|
||||
responses: {
|
||||
/** @description Successful Response */
|
||||
200: {
|
||||
headers: {
|
||||
[name: string]: unknown;
|
||||
};
|
||||
content: {
|
||||
"application/json": unknown;
|
||||
};
|
||||
};
|
||||
/** @description Validation Error */
|
||||
422: {
|
||||
headers: {
|
||||
[name: string]: unknown;
|
||||
};
|
||||
content: {
|
||||
"application/json": components["schemas"]["HTTPValidationError"];
|
||||
};
|
||||
};
|
||||
};
|
||||
};
|
||||
typesafe_proxy_route_typesafe__endpoint__get: {
|
||||
parameters: {
|
||||
query?: never;
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue