mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-16 04:29:31 +00:00
refactor: clean up fresh tech debt in per-user Copilot auth, Ternary FOCUS export and telemetry (#45769)
* refactor: read typed credential fields directly in per-user Copilot auth
- github_copilot_auth_mode reads credential.credential_values directly instead of getattr+isinstance+cast
- _new_async_github_client uses plain attribute access for get_async_httpx_client
- _per_user_oauth_configured and _all_per_user_credential_names iterate typed CredentialItem fields
- list_user_connections builds login/connected_at maps in one shot instead of seeded mutable dicts
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
* refactor: read typed fields directly in team member budget auth and pass-through route registration
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
* Revert "refactor: read typed fields directly in team member budget auth and pass-through route registration"
This reverts commit ae37be318d.
* refactor: deduplicate Ternary export locking and type HTTP handler override
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
* refactor: drop dead telemetry helpers, reuse PrismaClient.writer_db, remove restating comments
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
* ci: retrigger unit shards after an unrelated mcp-elicitation timeout
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---------
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
parent
23025c92ba
commit
1311894845
13 changed files with 65 additions and 98 deletions
|
|
@ -9,12 +9,14 @@ from urllib.parse import quote, urlparse
|
|||
from typing_extensions import NotRequired, ReadOnly, TypedDict
|
||||
|
||||
from litellm._logging import verbose_logger
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPEventHooks
|
||||
from litellm.types.llms.custom_http import VerifyTypes
|
||||
|
||||
from .base import FocusDestination, FocusTimeWindow
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import httpx
|
||||
from aiohttp import ClientSession
|
||||
|
||||
TERNARY_UPLOAD_TIMEOUT_SECONDS: Final = 120.0
|
||||
|
||||
|
|
@ -57,10 +59,17 @@ class _NoRedirectHTTPHandler(AsyncHTTPHandler):
|
|||
|
||||
def create_client(
|
||||
self,
|
||||
*args: object,
|
||||
**kwargs: object, # kwargs-ok: passthrough to AsyncHTTPHandler.create_client
|
||||
timeout: float | httpx.Timeout | None,
|
||||
event_hooks: HTTPEventHooks | None,
|
||||
ssl_verify: VerifyTypes | None = None,
|
||||
shared_session: ClientSession | None = None,
|
||||
) -> httpx.AsyncClient:
|
||||
client: Final = super().create_client(*args, **kwargs) # pyright: ignore[reportArgumentType] # passthrough
|
||||
client: Final = super().create_client(
|
||||
timeout=timeout,
|
||||
event_hooks=event_hooks,
|
||||
ssl_verify=ssl_verify,
|
||||
shared_session=shared_session,
|
||||
)
|
||||
client.follow_redirects = False
|
||||
return client
|
||||
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ from __future__ import annotations
|
|||
|
||||
import os
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import TYPE_CHECKING, Any, Final, cast
|
||||
from typing import TYPE_CHECKING, Any, ClassVar, Final, cast
|
||||
|
||||
import litellm
|
||||
from litellm._logging import verbose_logger
|
||||
|
|
@ -28,6 +28,8 @@ DEFAULT_DRY_RUN_LIMIT: Final = 500
|
|||
class FocusLogger(CustomLogger):
|
||||
"""Coordinates Focus export jobs across transformer/serializer/destination layers."""
|
||||
|
||||
export_job_name: ClassVar[str] = FOCUS_USAGE_DATA_JOB_NAME
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
|
|
@ -120,14 +122,14 @@ class FocusLogger(CustomLogger):
|
|||
pod_lock_manager = getattr(writer, "pod_lock_manager", None)
|
||||
|
||||
if pod_lock_manager and pod_lock_manager.redis_cache:
|
||||
acquired: Final = await pod_lock_manager.acquire_lock(cronjob_id=FOCUS_USAGE_DATA_JOB_NAME)
|
||||
acquired: Final = await pod_lock_manager.acquire_lock(cronjob_id=self.export_job_name)
|
||||
if not acquired:
|
||||
verbose_logger.debug("Focus export: unable to acquire pod lock")
|
||||
return
|
||||
try:
|
||||
await self._run_scheduled_export()
|
||||
finally:
|
||||
await pod_lock_manager.release_lock(cronjob_id=FOCUS_USAGE_DATA_JOB_NAME)
|
||||
await pod_lock_manager.release_lock(cronjob_id=self.export_job_name)
|
||||
else:
|
||||
await self._run_scheduled_export()
|
||||
|
||||
|
|
|
|||
|
|
@ -222,16 +222,24 @@ class TernaryExportEngine(FocusExportEngine):
|
|||
if data.is_empty():
|
||||
verbose_logger.debug("Ternary export: no usage data for window %s", window)
|
||||
return
|
||||
failed: Final[list[str]] = [] # mutable-ok: days whose upload failed, reported once after all days run
|
||||
for day, day_data in _partition_by_day(data):
|
||||
try:
|
||||
await self._deliver_day(day_data, _day_window(day, window.frequency))
|
||||
except Exception as e: # noqa: BLE001 # one day's failure must not block the rest; logged and re-raised
|
||||
verbose_logger.warning("Ternary export: upload for %s failed: %s", day, e)
|
||||
failed.append(day)
|
||||
failed: Final = tuple(
|
||||
[
|
||||
day
|
||||
for day, day_data in _partition_by_day(data)
|
||||
if not await TernaryExportEngine._deliver_day_or_warn(self, day, day_data, window.frequency)
|
||||
]
|
||||
)
|
||||
if failed:
|
||||
raise RuntimeError(f"Ternary export: {len(failed)} day(s) failed to upload: {', '.join(failed)}")
|
||||
|
||||
async def _deliver_day_or_warn(self, day: str, day_data: _Frame, frequency: str) -> bool:
|
||||
try:
|
||||
await self._deliver_day(day_data, _day_window(day, frequency))
|
||||
except Exception as e: # noqa: BLE001 # one day's failure must not block the rest; logged and re-raised
|
||||
verbose_logger.warning("Ternary export: upload for %s failed: %s", day, e)
|
||||
return False
|
||||
return True
|
||||
|
||||
async def _deliver_day(self, data: _Frame, window: FocusTimeWindow) -> None:
|
||||
payload: Final = _day_payload(data, transformer=self._transformer, serializer=self._serializer)
|
||||
if not payload:
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ from __future__ import annotations
|
|||
import os
|
||||
from collections.abc import Awaitable, Callable
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import TYPE_CHECKING, Final, Protocol
|
||||
from typing import TYPE_CHECKING, ClassVar, Final, Protocol
|
||||
|
||||
import litellm
|
||||
from litellm._logging import verbose_logger
|
||||
|
|
@ -51,6 +51,8 @@ class TernaryLogger(FocusLogger):
|
|||
TERNARY_EXPORT_FREQUENCY ("daily" default / "interval"), TERNARY_EXPORT_INTERVAL_SECONDS.
|
||||
"""
|
||||
|
||||
export_job_name: ClassVar[str] = TERNARY_USAGE_DATA_JOB_NAME
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
|
|
@ -80,13 +82,15 @@ class TernaryLogger(FocusLogger):
|
|||
"TERNARY_EXPORT_INTERVAL_SECONDS must be a positive integer when TERNARY_EXPORT_FREQUENCY is 'interval'"
|
||||
)
|
||||
|
||||
destination_config: Final[dict[str, str]] = {} # mutable-ok: built from the config values present below
|
||||
if resolved_api_key:
|
||||
destination_config["api_key"] = resolved_api_key
|
||||
if resolved_connection_id:
|
||||
destination_config["connection_id"] = resolved_connection_id
|
||||
if resolved_base_url:
|
||||
destination_config["base_url"] = resolved_base_url
|
||||
destination_config: Final = {
|
||||
key: value
|
||||
for key, value in (
|
||||
("api_key", resolved_api_key),
|
||||
("connection_id", resolved_connection_id),
|
||||
("base_url", resolved_base_url),
|
||||
)
|
||||
if value
|
||||
}
|
||||
|
||||
super().__init__(
|
||||
provider="ternary",
|
||||
|
|
@ -123,25 +127,6 @@ class TernaryLogger(FocusLogger):
|
|||
)
|
||||
return self._engine
|
||||
|
||||
async def initialize_focus_export_job(self) -> None:
|
||||
"""Run the scheduled export under a Ternary-specific pod lock (distinct from FocusLogger's)."""
|
||||
from litellm.proxy.proxy_server import proxy_logging_obj
|
||||
|
||||
writer: Final = getattr(proxy_logging_obj, "db_spend_update_writer", None) if proxy_logging_obj else None
|
||||
pod_lock_manager: Final = getattr(writer, "pod_lock_manager", None) if writer is not None else None
|
||||
|
||||
if pod_lock_manager and pod_lock_manager.redis_cache:
|
||||
acquired: Final = await pod_lock_manager.acquire_lock(cronjob_id=TERNARY_USAGE_DATA_JOB_NAME)
|
||||
if not acquired:
|
||||
verbose_logger.debug("Ternary export: unable to acquire pod lock")
|
||||
return
|
||||
try:
|
||||
await self._run_scheduled_export()
|
||||
finally:
|
||||
await pod_lock_manager.release_lock(cronjob_id=TERNARY_USAGE_DATA_JOB_NAME)
|
||||
else:
|
||||
await self._run_scheduled_export()
|
||||
|
||||
@staticmethod
|
||||
async def init_ternary_background_job(
|
||||
scheduler: _JobScheduler,
|
||||
|
|
|
|||
|
|
@ -74,6 +74,8 @@ _IPV4_LOCAL_ADDRESS: Final = "0.0.0.0"
|
|||
|
||||
_HttpxTransportT = TypeVar("_HttpxTransportT", HTTPTransport, AsyncHTTPTransport)
|
||||
|
||||
HTTPEventHooks: TypeAlias = Mapping[str, list[Callable[..., object]]]
|
||||
|
||||
|
||||
def http2_enabled() -> bool:
|
||||
from litellm.secret_managers.main import str_to_bool
|
||||
|
|
@ -666,7 +668,7 @@ class AsyncHTTPHandler:
|
|||
def create_client(
|
||||
self,
|
||||
timeout: float | httpx.Timeout | None,
|
||||
event_hooks: Mapping[str, list[Callable[..., object]]] | None,
|
||||
event_hooks: HTTPEventHooks | None,
|
||||
ssl_verify: VerifyTypes | None = None,
|
||||
shared_session: Optional["ClientSession"] = None,
|
||||
) -> httpx.AsyncClient:
|
||||
|
|
|
|||
|
|
@ -328,7 +328,7 @@ def _new_async_github_client() -> _AsyncGitHubClient:
|
|||
|
||||
return cast( # cast-ok: pins the untyped litellm client factory to the local Protocol
|
||||
Callable[[str], _AsyncGitHubClient],
|
||||
getattr(http_handler, "get_async_httpx_client"), # noqa: B009 # untyped factory
|
||||
http_handler.get_async_httpx_client,
|
||||
)(_LLM_PROVIDER)
|
||||
|
||||
|
||||
|
|
@ -548,13 +548,7 @@ def github_copilot_auth_mode(litellm_credential_name: object, auth_type_value: o
|
|||
|
||||
credential: Final = CredentialAccessor.find_credential(litellm_credential_name)
|
||||
if credential is not None:
|
||||
values: Final[object] = getattr(credential, "credential_values", None)
|
||||
return isinstance(values, Mapping) and (
|
||||
cast( # cast-ok: value is Mapping-checked or dict-shaped at runtime
|
||||
Mapping[object, object], values
|
||||
).get(GITHUB_COPILOT_AUTH_TYPE_KEY)
|
||||
== GITHUB_COPILOT_PER_USER_AUTH_TYPE
|
||||
)
|
||||
return credential.credential_values.get(GITHUB_COPILOT_AUTH_TYPE_KEY) == GITHUB_COPILOT_PER_USER_AUTH_TYPE
|
||||
return auth_type_value == GITHUB_COPILOT_PER_USER_AUTH_TYPE
|
||||
if auth_type_value == GITHUB_COPILOT_PER_USER_AUTH_TYPE:
|
||||
raise BadRequestError(
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@ CRUD endpoints for storing reusable credentials.
|
|||
import time
|
||||
import unicodedata
|
||||
from collections.abc import Mapping
|
||||
from datetime import datetime
|
||||
from typing import (
|
||||
Annotated,
|
||||
Final,
|
||||
|
|
@ -563,17 +562,12 @@ async def list_user_connections(
|
|||
detail={"error": CommonProxyErrors.db_not_connected_error.value},
|
||||
)
|
||||
rows: Final = await list_user_provider_credentials(prisma_client, user_id)
|
||||
github_logins: Final[dict[str, str]] = {} # mutable-ok: accumulates one entry per credential row
|
||||
connected_at: Final[dict[str, str]] = {} # mutable-ok: accumulates one entry per credential row
|
||||
for row in rows:
|
||||
if row.provider != _GITHUB_COPILOT_PROVIDER:
|
||||
continue
|
||||
payload = decode_user_provider_credential(row.credential_b64)
|
||||
if payload is not None:
|
||||
github_logins[row.credential_name] = payload.github_login
|
||||
updated = getattr(row, "updated_at", None)
|
||||
if isinstance(updated, datetime):
|
||||
connected_at[row.credential_name] = updated.isoformat()
|
||||
copilot_rows: Final = tuple(row for row in rows if row.provider == _GITHUB_COPILOT_PROVIDER)
|
||||
payloads: Final = {
|
||||
row.credential_name: decode_user_provider_credential(row.credential_b64) for row in copilot_rows
|
||||
}
|
||||
github_logins: Final = {name: payload.github_login for name, payload in payloads.items() if payload is not None}
|
||||
connected_at: Final = {row.credential_name: row.updated_at.isoformat() for row in copilot_rows}
|
||||
return UserProviderConnectionsResponse(
|
||||
connections=[
|
||||
UserProviderConnection(
|
||||
|
|
|
|||
|
|
@ -2732,11 +2732,7 @@ def _per_user_oauth_configured() -> bool:
|
|||
)
|
||||
|
||||
return any(
|
||||
isinstance(values := getattr(credential, "credential_values", None), Mapping)
|
||||
and cast("Mapping[object, object]", values).get( # cast-ok: credential_values is a plain dict at runtime
|
||||
GITHUB_COPILOT_AUTH_TYPE_KEY
|
||||
)
|
||||
== GITHUB_COPILOT_PER_USER_AUTH_TYPE
|
||||
credential.credential_values.get(GITHUB_COPILOT_AUTH_TYPE_KEY) == GITHUB_COPILOT_PER_USER_AUTH_TYPE
|
||||
for credential in (litellm.credential_list or ())
|
||||
)
|
||||
|
||||
|
|
@ -2753,12 +2749,8 @@ def _all_per_user_credential_names() -> tuple[str, ...]:
|
|||
return tuple(
|
||||
credential.credential_name
|
||||
for credential in (litellm.credential_list or ())
|
||||
if getattr(credential, "credential_name", None)
|
||||
and isinstance(values := getattr(credential, "credential_values", None), Mapping)
|
||||
and cast("Mapping[object, object]", values).get( # cast-ok: credential_values is a plain dict at runtime
|
||||
GITHUB_COPILOT_AUTH_TYPE_KEY
|
||||
)
|
||||
== GITHUB_COPILOT_PER_USER_AUTH_TYPE
|
||||
if credential.credential_name
|
||||
and credential.credential_values.get(GITHUB_COPILOT_AUTH_TYPE_KEY) == GITHUB_COPILOT_PER_USER_AUTH_TYPE
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1717,7 +1717,7 @@ async def proxy_startup_event(app: FastAPI) -> AsyncGenerator[ProxyLifespanState
|
|||
litellm_version=version,
|
||||
settings=telemetry_settings,
|
||||
db=lambda: ( # pyright: ignore[reportArgumentType] # PrismaWrapper forwards raw queries via __getattr__
|
||||
getattr(prisma_client.db, "writer", prisma_client.db) if prisma_client is not None else None
|
||||
prisma_client.writer_db if prisma_client is not None else None
|
||||
),
|
||||
register=litellm.logging_callback_manager.add_litellm_callback,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -48,7 +48,7 @@ def telemetry_store() -> TelemetryStore | None:
|
|||
if known is not None or prisma_client is None:
|
||||
return known
|
||||
return TelemetryStore(
|
||||
getattr(prisma_client.db, "writer", prisma_client.db), # pyright: ignore[reportArgumentType] # PrismaWrapper forwards raw queries via __getattr__
|
||||
prisma_client.writer_db, # pyright: ignore[reportArgumentType] # PrismaWrapper forwards raw queries via __getattr__
|
||||
telemetry_runtime.settings.retention_days,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ import uuid
|
|||
from collections.abc import Awaitable, Callable, Coroutine, Iterator, Mapping
|
||||
from dataclasses import dataclass
|
||||
from types import MappingProxyType
|
||||
from typing import Final, TypeAlias
|
||||
from typing import Final
|
||||
|
||||
import httpx
|
||||
|
||||
|
|
@ -33,8 +33,6 @@ from litellm.telemetry.report import Report
|
|||
from litellm.telemetry.sink import Exporter, ExportOutcome
|
||||
from litellm.types.llms.custom_http import httpxSpecialProvider
|
||||
|
||||
StoredConsent: TypeAlias = Callable[[], Awaitable[TelemetryConsent | None]]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class StoredChoice:
|
||||
|
|
|
|||
|
|
@ -9312,11 +9312,11 @@ def validate_model_access(
|
|||
_PRESERVED_NONE_FIELDS: Final[list[tuple[str, str]]] = [
|
||||
("message", "content"), # null when tool_calls present (issue #6677)
|
||||
("message", "role"), # always required by OpenAI spec
|
||||
("message", "refusal"), # explicit null from provider
|
||||
("message", "refusal"),
|
||||
("delta", "content"), # null in streaming chunks
|
||||
]
|
||||
_PRESERVED_NONE_CHOICE_FIELDS: Final = ("logprobs",) # explicit null from provider
|
||||
_PRESERVED_NONE_RESPONSE_FIELDS: Final = ("system_fingerprint",) # explicit null from provider
|
||||
_PRESERVED_NONE_CHOICE_FIELDS: Final = ("logprobs",)
|
||||
_PRESERVED_NONE_RESPONSE_FIELDS: Final = ("system_fingerprint",)
|
||||
|
||||
|
||||
def model_dump_with_preserved_fields(
|
||||
|
|
|
|||
|
|
@ -17,23 +17,6 @@ class TelemetrySink(Protocol):
|
|||
async def flush(self) -> None: ...
|
||||
|
||||
|
||||
class NoopSink:
|
||||
def set_instance(self, info: InstanceInfo) -> None:
|
||||
return None
|
||||
|
||||
def record_request(self, record: RequestRecord) -> None:
|
||||
return None
|
||||
|
||||
def record_attempt(self, record: AttemptRecord) -> None:
|
||||
return None
|
||||
|
||||
def record_ui_event(self, event: UIEvent) -> None:
|
||||
return None
|
||||
|
||||
async def flush(self) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class ExportOutcome(str, Enum):
|
||||
SENT = "sent"
|
||||
REJECTED = "rejected"
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue