scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-13 23:11:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 152

refactor(otel/v2): satisfy the type-discipline and strict-lint ceilings

Browse source 
The stack was rebased onto a base whose ceilings had since been ratcheted down, so code that fit before no longer did. Never-rebound names carry Final, mapping payloads are built immutably through MappingProxyType, the four duplicated access-rejection bodies collapse into one helper, and the listing's optional verdict key moves into a named function.

Four constructions keep a mutable-ok with a reason: each is handed to FastAPI or an OTLP exporter, which need a concrete dict.
This commit is contained in:
Yucheng Zhu 2026-08-08 11:55:03 -07:00
parent 29a0068763
commit 89eee4098f
4 changed files with 86 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

81
litellm/integrations/otel/presets/destinations.py
View file

@ -12,6 +12,8 @@ credential values only.
import os
from collections.abc import Callable, Mapping
from types import MappingProxyType
from typing import Final
from litellm.integrations.langfuse.langfuse_otel import (
LANGFUSE_CLOUD_US_ENDPOINT,
@ -22,48 +24,52 @@ from litellm.integrations.weave.weave_otel import _get_weave_authorization_heade
def _parse_header_string(raw: str) -> Mapping[str, str]:
pairs = (item.split("=", 1) for item in raw.split(",") if "=" in item)
return {key.strip(): value.strip() for key, value in pairs}
pairs: Final = (item.split("=", 1) for item in raw.split(",") if "=" in item)
return MappingProxyType({key.strip(): value.strip() for key, value in pairs})
def _langfuse_endpoint(host: str) -> str:
normalized = host if host.startswith("http") else f"https://{host}"
normalized: Final = host if host.startswith("http") else f"https://{host}"
return f"{normalized.rstrip('/')}/api/public/otel"
def _langfuse_destination(values: Mapping[str, str]) -> OtelDestination | None:
public_key = values.get("langfuse_public_key")
secret_key = values.get("langfuse_secret_key")
public_key: Final = values.get("langfuse_public_key")
secret_key: Final = values.get("langfuse_secret_key")
if not public_key or not secret_key:
return None
host = values.get("langfuse_host")
endpoint = _langfuse_endpoint(host) if host else LANGFUSE_CLOUD_US_ENDPOINT
auth = LangfuseOtelLogger._get_langfuse_authorization_header(public_key=public_key, secret_key=secret_key)
return OtelDestination(endpoint=endpoint, headers={"Authorization": auth})
host: Final = values.get("langfuse_host")
endpoint: Final = _langfuse_endpoint(host) if host else LANGFUSE_CLOUD_US_ENDPOINT
auth: Final = LangfuseOtelLogger._get_langfuse_authorization_header(public_key=public_key, secret_key=secret_key)
return OtelDestination(endpoint=endpoint, headers=MappingProxyType({"Authorization": auth}))
def _arize_destination(values: Mapping[str, str]) -> OtelDestination | None:
space = values.get("arize_space_id") or values.get("arize_space_key")
api_key = values.get("arize_api_key")
space: Final = values.get("arize_space_id") or values.get("arize_space_key")
api_key: Final = values.get("arize_api_key")
if not space or not api_key:
return None
# Mirrors the global config's ARIZE_ENDPOINT / ARIZE_HTTP_ENDPOINT split: Arize's
# own endpoint is gRPC, but a destination may point at an HTTP collector, and the
# URL scheme cannot express that (the gRPC endpoint is also https://).
http_endpoint = values.get("arize_http_endpoint")
endpoint = values.get("arize_endpoint") or http_endpoint or "https://otlp.arize.com/v1"
project = values.get("arize_project_name") or values.get("project_name") or os.environ.get("ARIZE_PROJECT_NAME")
resource_attributes = {"model_id": project, "arize.project.name": project} if project else {}
http_endpoint: Final = values.get("arize_http_endpoint")
endpoint: Final = values.get("arize_endpoint") or http_endpoint or "https://otlp.arize.com/v1"
project: Final = (
values.get("arize_project_name") or values.get("project_name") or os.environ.get("ARIZE_PROJECT_NAME")
)
resource_attributes: Final = (
MappingProxyType({"model_id": project, "arize.project.name": project}) if project else _EMPTY
)
return OtelDestination(
endpoint=endpoint,
headers={"space_id": space, "api_key": api_key},
headers=MappingProxyType({"space_id": space, "api_key": api_key}),
resource_attributes=resource_attributes,
protocol="otlp_http" if http_endpoint and not values.get("arize_endpoint") else None,
)
def _weave_destination(values: Mapping[str, str]) -> OtelDestination | None:
api_key = values.get("wandb_api_key")
api_key: Final = values.get("wandb_api_key")
if not api_key:
return None
from litellm.integrations.weave.weave_otel import (
@ -71,12 +77,15 @@ def _weave_destination(values: Mapping[str, str]) -> OtelDestination | None:
WEAVE_OTEL_ENDPOINT,
)
base = (values.get("weave_endpoint") or WEAVE_BASE_URL).rstrip("/")
endpoint = base if base.endswith("/v1/traces") else base.removesuffix("/otel") + WEAVE_OTEL_ENDPOINT
headers = {"Authorization": _get_weave_authorization_header(api_key=api_key)}
project_id = values.get("weave_project_id")
if project_id:
headers["project_id"] = project_id
base: Final = (values.get("weave_endpoint") or WEAVE_BASE_URL).rstrip("/")
endpoint: Final = base if base.endswith("/v1/traces") else base.removesuffix("/otel") + WEAVE_OTEL_ENDPOINT
project_id: Final = values.get("weave_project_id")
headers: Final = MappingProxyType(
{ # mutable-ok: the OTLP exporter is handed a concrete header map
"Authorization": _get_weave_authorization_header(api_key=api_key),
**({"project_id": project_id} if project_id else {}), # mutable-ok: optional key, spread inline
}
)
return OtelDestination(endpoint=endpoint, headers=headers)
@ -91,7 +100,7 @@ def _generic_destination(values: Mapping[str, str]) -> OtelDestination | None:
to the plain HTTP URL the admin typed, so the destination silently delivered nothing
while still being disclosed as active.
"""
endpoint = values.get("otel_endpoint")
endpoint: Final = values.get("otel_endpoint")
if not endpoint:
return None
return OtelDestination(
@ -101,13 +110,17 @@ def _generic_destination(values: Mapping[str, str]) -> OtelDestination | None:
)
_ADAPTERS: Mapping[str, Callable[[Mapping[str, str]], OtelDestination | None]] = {
"langfuse_otel": _langfuse_destination,
"arize": _arize_destination,
"weave_otel": _weave_destination,
}
_EMPTY: Final[Mapping[str, str]] = MappingProxyType({})
OTEL_V2_DESTINATION_CALLBACKS = frozenset(_ADAPTERS)
_ADAPTERS: Final[Mapping[str, Callable[[Mapping[str, str]], OtelDestination | None]]] = MappingProxyType(
{
"langfuse_otel": _langfuse_destination,
"arize": _arize_destination,
"weave_otel": _weave_destination,
}
)
OTEL_V2_DESTINATION_CALLBACKS: Final = frozenset(_ADAPTERS)
def build_destination(callback_name: str, values: Mapping[str, str]) -> OtelDestination | None:
@ -118,10 +131,12 @@ def build_destination(callback_name: str, values: Mapping[str, str]) -> OtelDest
host (an easy slip in the create form) yields a malformed OTLP URL the
exporter rejects with a 404, so whitespace is never significant here.
"""
trimmed = {key: value.strip() if isinstance(value, str) else value for key, value in values.items()}
adapter = _ADAPTERS.get(callback_name)
trimmed: Final = MappingProxyType(
{key: value.strip() for key, value in values.items()}
)
adapter: Final = _ADAPTERS.get(callback_name)
if adapter is not None:
destination = adapter(trimmed)
destination: Final = adapter(trimmed)
if destination is not None:
return destination
return _generic_destination(trimmed)

18
litellm/proxy/credential_endpoints/endpoints.py
View file

@ -2,6 +2,8 @@
CRUD endpoints for storing reusable credentials.
"""
from collections.abc import Mapping
from types import MappingProxyType
from typing import Final
from fastapi import APIRouter, Depends, HTTPException, Path, Request, Response
@ -26,6 +28,8 @@ from litellm.types.utils import CreateCredentialItem, CredentialItem
router: Final = APIRouter()
_EMPTY_VERDICT: Final[Mapping[str, bool]] = MappingProxyType({})
class CredentialHelperUtils:
@staticmethod
@ -121,6 +125,14 @@ async def create_credential(
dependencies=[Depends(user_api_key_auth)],
tags=["credential management"],
)
def _destination_verdict(credential: CredentialItem) -> Mapping[str, bool]:
"""Whether this logging credential actually builds, for the listing. Empty for a
provider credential, which has no destination to report on."""
if not is_logging_credential(credential.credential_info):
return _EMPTY_VERDICT
return {"resolves_to_destination": destination_for_credential(credential) is not None} # mutable-ok: JSON body
async def get_credentials(
request: Request,
fastapi_response: Response,
@ -135,11 +147,7 @@ async def get_credentials(
"credential_name": credential.credential_name,
"credential_values": _get_masked_values(credential.credential_values),
"credential_info": credential.credential_info,
**(
{"resolves_to_destination": destination_for_credential(credential) is not None}
if is_logging_credential(credential.credential_info)
else {}
),
**_destination_verdict(credential),
}
for credential in litellm.credential_list
]

15
litellm/proxy/management_endpoints/logging_exporter_access.py
View file

@ -10,7 +10,9 @@ scope is the given set of team ids and org ids. The resolver passes a
one-element scope built with ``identity_scope``.
"""
from typing import TYPE_CHECKING
from collections.abc import Mapping
from types import MappingProxyType
from typing import TYPE_CHECKING, Final
from pydantic import BaseModel, ConfigDict, ValidationError
@ -22,6 +24,9 @@ if TYPE_CHECKING:
from litellm.integrations.otel.model.destination import OtelDestination
_EMPTY: Final[Mapping[str, object]] = MappingProxyType({})
class _LoggingDestinationTag(BaseModel):
"""Lenient read of just the ``credential_type`` tag, ignoring the rest of
``credential_info`` (including a possibly-malformed ``access``)."""
@ -127,14 +132,16 @@ def destination_for_credential(credential: CredentialItem) -> 'tuple[str, "OtelD
from litellm.integrations.otel.presets import PRESET_BY_CALLBACK
from litellm.integrations.otel.presets.destinations import build_destination
backend = (credential.credential_info or {}).get("description")
backend: Final = (credential.credential_info or _EMPTY).get("description")
if not backend:
return None
# Drop unset (``None``) values rather than stringifying them: ``str(None)`` is the
# literal ``"None"``, which would land in the exporter endpoint/headers and break
# the export (e.g. an empty ``otel_endpoint`` becoming the URL ``"None"``).
values = {str(key): str(value) for key, value in (credential.credential_values or {}).items() if value is not None}
destination = build_destination(backend, values)
values: Final = MappingProxyType(
{str(key): str(value) for key, value in (credential.credential_values or _EMPTY).items() if value is not None}
)
destination: Final = build_destination(backend, values)
if destination is None:
return None
return (backend if backend in PRESET_BY_CALLBACK else "generic", destination)

32
litellm/proxy/management_endpoints/logging_exporter_validation.py
View file

@ -6,9 +6,17 @@ request time. This module only checks that a write sets a well-formed ``access``
"""
from collections.abc import Mapping
from typing import Final, NoReturn
from fastapi import HTTPException, status
_ALLOWED_ACCESS_FIELDS: Final = frozenset({"global", "teams", "orgs"})
def _reject(message: str) -> NoReturn:
detail: Final = {"error": message} # mutable-ok: FastAPI serialises the detail
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=detail)
def validate_credential_access(credential_info: Mapping[str, object] | None) -> None:
"""Validate ``credential_info.access`` shape when the write sets one.
@ -19,27 +27,15 @@ def validate_credential_access(credential_info: Mapping[str, object] | None) ->
"""
if not isinstance(credential_info, dict) or "access" not in credential_info:
return
access = credential_info["access"]
access: Final = credential_info["access"]
if not isinstance(access, dict):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": "credential_info.access must be an object"},
)
_reject("credential_info.access must be an object")
if "global" in access and not isinstance(access["global"], bool):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": "access.global must be a boolean"},
)
_reject("access.global must be a boolean")
for field in ("teams", "orgs"):
bucket = access.get(field)
if bucket is not None and not (isinstance(bucket, list) and all(isinstance(item, str) for item in bucket)):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": f"access.{field} must be a list of strings"},
)
unknown = set(access) - {"global", "teams", "orgs"}
_reject(f"access.{field} must be a list of strings")
unknown: Final = frozenset(access) - _ALLOWED_ACCESS_FIELDS
if unknown:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"error": f"access contains unknown field(s): {sorted(unknown)}"},
)
_reject(f"access contains unknown field(s): {sorted(unknown)}")