diff --git a/litellm/ocr/main.py b/litellm/ocr/main.py index a4db8c5b0e5..44dd6210aaf 100644 --- a/litellm/ocr/main.py +++ b/litellm/ocr/main.py @@ -38,4 +38,3 @@ async def aocr(*args: object, **kwargs: object) -> OCRResponse: # kwargs-ok: pr Callable[..., Awaitable[OCRResponse]], legacy.aocr ) return await fallback(*args, **kwargs) - diff --git a/litellm/rust_bridge/leaves.py b/litellm/rust_bridge/leaves.py index 64e239d1615..6b9020aba78 100644 --- a/litellm/rust_bridge/leaves.py +++ b/litellm/rust_bridge/leaves.py @@ -17,12 +17,14 @@ import datetime import json import traceback from collections.abc import Awaitable, Callable, Mapping -from typing import ( # noqa: TID251 # narrows the untyped legacy Logging and CustomLogger surfaces once +from types import MappingProxyType +from typing import ( TYPE_CHECKING, Final, Literal, Protocol, - cast, + TypeAlias, + cast, # noqa: TID251 # narrows the untyped legacy Logging and CustomLogger surfaces once ) from litellm.integrations.custom_logger import CustomLogger @@ -30,12 +32,14 @@ from litellm.integrations.custom_logger import CustomLogger if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging -TerminalFamily = Literal["sync_success", "async_success", "sync_failure", "async_failure"] -Family = Literal["request", TerminalFamily] -Details = dict[str, object] -Timestamp = datetime.datetime -LegacyCall = Callable[..., object] -LegacyAsyncCall = Callable[..., Awaitable[None]] +TerminalFamily: TypeAlias = Literal["sync_success", "async_success", "sync_failure", "async_failure"] +Family: TypeAlias = Literal["request", TerminalFamily] +Details: TypeAlias = dict[ + str, object +] # mutable-ok: model_call_details is the shared mutable envelope callbacks write to +Timestamp: TypeAlias = datetime.datetime +LegacyCall: TypeAlias = Callable[..., object] +LegacyAsyncCall: TypeAlias = Callable[..., Awaitable[None]] class LoggerView(Protocol): @@ -60,17 +64,19 @@ class LoggerView(Protocol): def _pre_call(self, input: str, api_key: str | None, model: str | None, additional_args: Details) -> None: ... - def _print_llm_call_debugging_log(self, api_base: str, headers: Details, additional_args: Details) -> None: ... + def _print_llm_call_debugging_log( + self, api_base: str, headers: Mapping[str, str], additional_args: Details + ) -> None: ... def _get_request_curl_command( - self, api_base: str, headers: Details | None, additional_args: Details, data: object + self, api_base: str, headers: Mapping[str, str] | None, additional_args: Details, data: object ) -> str: ... def _get_masked_api_base(self, api_base: str) -> str: ... def _get_raw_request_body(self, data: object) -> Details: ... - def _get_masked_headers(self, headers: Details) -> Details: ... + def _get_masked_headers(self, headers: Mapping[str, str]) -> Details: ... def _response_cost_calculator(self, result: object) -> float | None: ... @@ -137,6 +143,31 @@ class IntegrationView(Protocol): ) -> Awaitable[None]: ... +_EMPTY: Final[Mapping[str, object]] = MappingProxyType({}) + + +def _details_of(value: object) -> Details: + return cast(Details, value) # cast-ok: legacy model_call_details nesting is untyped and shared by callbacks + + +def _call_of(value: object) -> LegacyCall: + return cast(LegacyCall, value) # cast-ok: legacy integration entry points are untyped + + +def _attribute(target: object, name: str) -> object: + value: Final[object] = getattr(target, name) # pyright: ignore[reportAny] # legacy attributes are untyped by definition + return value + + +def _seed(details: Details, key: str) -> Details: + existing: Final[object] = details.get(key) + if isinstance(existing, dict): + return _details_of(existing) # pyright: ignore[reportUnknownArgumentType] # isinstance yields dict[Unknown, Unknown] + seeded: Final[Details] = {} # mutable-ok: legacy envelope seed shared with callbacks + details[key] = seeded # rebind-ok: seeding a nested dict on the shared envelope is the legacy contract + return seeded + + def _logger(logger: Logging) -> LoggerView: return cast(LoggerView, logger) # cast-ok: legacy Logging is untyped; this protocol names the attributes we read @@ -148,9 +179,9 @@ def _integration(callback: CustomLogger) -> IntegrationView: def _legacy_module() -> Mapping[str, object]: from litellm.litellm_core_utils import litellm_logging - return cast( + return cast( # cast-ok: module globals hold the legacy integration singletons Mapping[str, object], vars(litellm_logging) - ) # cast-ok: module globals hold the legacy integration singletons + ) def _print_verbose() -> LegacyCall: @@ -160,17 +191,18 @@ def _print_verbose() -> LegacyCall: def _method(target: object, name: str) -> LegacyCall: - return cast(LegacyCall, getattr(target, name)) # cast-ok: legacy integration singletons are untyped + return _call_of(_attribute(target, name)) def _async_method(target: object, name: str) -> LegacyAsyncCall: - return cast(LegacyAsyncCall, getattr(target, name)) # cast-ok: legacy integration singletons are untyped + return cast(LegacyAsyncCall, _attribute(target, name)) # cast-ok: legacy integration singletons are untyped def _redact_string(value: str) -> str: from litellm.litellm_core_utils import litellm_logging - return cast(Callable[[str], str], litellm_logging._redact_string)(value) # pyright: ignore[reportPrivateUsage] # cast-ok: legacy helper + redact: Final = _call_of(litellm_logging._redact_string) # pyright: ignore[reportPrivateUsage] # private legacy helper + return str(redact(value)) def _redact_result(details: Details, result: object) -> object: @@ -185,13 +217,17 @@ def record_pre_call( *, api_key: str | None, body: Details, - headers: dict[str, str], + headers: Mapping[str, str], url: str, ) -> None: view: Final = _logger(logger) - additional_args: Final[Details] = {"complete_input_dict": body, "headers": headers, "api_base": url} + additional_args: Final[Details] = { # mutable-ok: legacy additional_args is rebound by callbacks in place + "complete_input_dict": body, + "headers": headers, + "api_base": url, + } view._pre_call(input="OCR document processing", api_key=api_key, model=None, additional_args=additional_args) # pyright: ignore[reportPrivateUsage] # legacy state writer - view._print_llm_call_debugging_log(api_base=url, headers=dict(headers), additional_args=additional_args) # pyright: ignore[reportPrivateUsage] # legacy debug output + view._print_llm_call_debugging_log(api_base=url, headers=headers, additional_args=additional_args) # pyright: ignore[reportPrivateUsage] # legacy debug output _capture_raw_request(view, additional_args) _run_logger_fn(logger) view.record_api_call_start_time() @@ -204,15 +240,13 @@ def _capture_raw_request(view: LoggerView, additional_args: Details) -> None: if not (view.log_raw_request_response or litellm.log_raw_request_response): return details: Final = view.model_call_details - params: Final = cast(Details, details.get("litellm_params") or {}) # cast-ok: legacy nested dict - metadata: Final = cast(Details, params.get("metadata") or {}) # cast-ok: legacy nested dict - params.setdefault("metadata", metadata) + metadata: Final = _seed(_seed(details, "litellm_params"), "metadata") if litellm.turn_off_message_logging: metadata["raw_request"] = "redacted by litellm. 'litellm.turn_off_message_logging=True'" return api_base: Final = str(additional_args.get("api_base") or "") - headers: Final = cast(Details, additional_args.get("headers") or {}) # cast-ok: legacy nested dict - body: Final = additional_args.get("complete_input_dict", {}) + headers: Final = cast(Mapping[str, str], additional_args.get("headers") or _EMPTY) # cast-ok: legacy nested dict + body: Final = additional_args.get("complete_input_dict", _EMPTY) try: curl: Final = view._get_request_curl_command( # pyright: ignore[reportPrivateUsage] # legacy debug formatter api_base=api_base, headers=headers, additional_args=additional_args, data=body @@ -232,18 +266,16 @@ def _capture_raw_request(view: LoggerView, additional_args: Details) -> None: def _run_logger_fn(logger: Logging) -> None: from litellm._logging import verbose_logger - logger_fn: Final = cast( - Callable[[Details], object] | None, getattr(logger, "logger_fn", None) - ) # cast-ok: user hook is untyped + logger_fn: Final = getattr(logger, "logger_fn", None) if not callable(logger_fn): return try: - logger_fn(_logger(logger).model_call_details) + _call_of(logger_fn)(_logger(logger).model_call_details) except Exception as error: # noqa: BLE001 # user logger_fn failures never fail the request verbose_logger.exception("LiteLLM.LoggingError: [Non-Blocking] Exception occurred while logging %s", error) -def record_post_call(logger: Logging, *, original_response: object, body: Details, headers: dict[str, str]) -> None: +def record_post_call(logger: Logging, *, original_response: object, body: Details, headers: Mapping[str, str]) -> None: view: Final = _logger(logger) serialized: Final = ( json.dumps(original_response, default=str) if isinstance(original_response, dict) else original_response @@ -252,7 +284,7 @@ def record_post_call(logger: Logging, *, original_response: object, body: Detail original_response=serialized, input=None, api_key=None, - additional_args={"complete_input_dict": body, "headers": headers}, + additional_args={"complete_input_dict": body, "headers": headers}, # mutable-ok: rebound in place by callbacks ) _run_logger_fn(logger) _redact_result(view.model_call_details, serialized) @@ -330,12 +362,9 @@ def prepare_success_logging(logger: Logging, response: object, start_time: Times details["log_event_type"] = "successful_api_call" details["end_time"] = end_time details["cache_hit"] = None - hidden: Final = cast(Details, getattr(response, "_hidden_params", None) or {}) # cast-ok: legacy response attribute - params: Final = cast(Details | None, details.get("litellm_params")) # cast-ok: legacy nested dict - if hidden and params is not None: - metadata: Final = cast(Details, params.get("metadata") or {}) # cast-ok: legacy nested dict - params["metadata"] = metadata - metadata["hidden_params"] = hidden + hidden: Final = _details_of(getattr(response, "_hidden_params", None) or {}) # mutable-ok: legacy fallback envelope + if hidden and isinstance(details.get("litellm_params"), dict): + _seed(_seed(details, "litellm_params"), "metadata")["hidden_params"] = hidden existing: Final = details.get("response_cost") if "response_cost" in hidden: details["response_cost"] = hidden["response_cost"] @@ -345,7 +374,7 @@ def prepare_success_logging(logger: Logging, response: object, start_time: Times details["standard_logging_object"] = payload if payload is not None: emit_standard_logging_payload( - cast(StandardLoggingPayload, payload) + cast(StandardLoggingPayload, payload) # cast-ok: legacy builder returns the payload TypedDict ) # cast-ok: legacy builder returns the payload TypedDict return _redact_result(details, response) @@ -368,16 +397,16 @@ def prepare_failure_logging( if details.get("combined_usage_object") is None: details["response_cost"] = 0 headers: Final = getattr(exception, "headers", None) - if isinstance(headers, dict): - params: Final = cast(Details, details.setdefault("litellm_params", {})) # cast-ok: legacy nested dict - metadata: Final = cast(Details, params.get("metadata") or {}) # cast-ok: legacy nested dict - metadata.update(cast(Details, headers)) # cast-ok: exception headers are a plain dict - build: Final = cast( + if isinstance(headers, Mapping): + _seed(_seed(details, "litellm_params"), "metadata").update( + _details_of(headers) # pyright: ignore[reportUnknownArgumentType] # exception headers carry no element types + ) + build: Final = cast( # cast-ok: labeled leaf: native payload pending LegacyCall, litellm_logging.get_standard_logging_object_payload - ) # cast-ok: labeled leaf: native payload pending + ) details["standard_logging_object"] = build( kwargs=details, - init_response_obj={}, + init_response_obj=_EMPTY, start_time=start_time, end_time=end_time, logging_obj=logger, @@ -389,17 +418,19 @@ def prepare_failure_logging( return formatted -_EVENT_HOOKS: Final[Mapping[TerminalFamily, str]] = { - "sync_success": "success_handler", - "async_success": "async_success_handler", - "sync_failure": "failure_handler", - "async_failure": "async_failure_handler", -} +_EVENT_HOOKS: Final[Mapping[TerminalFamily, str]] = MappingProxyType( + { + "sync_success": "success_handler", + "async_success": "async_success_handler", + "sync_failure": "failure_handler", + "async_failure": "async_failure_handler", + } +) def should_run_callback(logger: Logging, callback: object, family: TerminalFamily) -> bool: view: Final = _logger(logger) - params: Final = cast(Details, view.model_call_details.get("litellm_params") or {}) # cast-ok: legacy nested dict + params: Final = _details_of(view.model_call_details.get("litellm_params") or _EMPTY) return view.should_run_callback(callback=callback, litellm_params=params, event_hook=_EVENT_HOOKS[family]) @@ -427,9 +458,9 @@ async def async_logging_hook(logger: Logging, callback: CustomLogger, result: ob from litellm.litellm_core_utils import redact_messages view: Final = _logger(logger) - redact: Final = cast( + redact: Final = cast( # cast-ok: legacy helper LegacyCall, redact_messages.redact_message_input_output_from_custom_logger - ) # cast-ok: legacy helper + ) redacted: Final = ( result if isinstance(callback, CustomGuardrail) @@ -467,9 +498,9 @@ def async_log_success_event( details: Final = integration.redact_standard_logging_payload_from_model_call_details( model_call_details=_logger(logger).model_call_details ) - redact: Final = cast( + redact: Final = cast( # cast-ok: legacy helper Callable[..., Details], redact_messages.redact_streaming_responses_for_custom_logger - ) # cast-ok: legacy helper + ) view: Final = redact(model_call_details=details, custom_logger=callback) return integration.async_log_success_event( kwargs=view, response_obj=response, start_time=start_time, end_time=end_time @@ -533,19 +564,21 @@ def dispatch_callable( ) -_SUCCESS_SINGLETONS: Final[Mapping[str, str]] = { - "promptlayer": "promptLayerLogger", - "supabase": "supabaseClient", - "wandb": "weightsBiasesLogger", - "logfire": "logfireLogger", - "lunary": "lunaryLogger", - "helicone": "heliconeLogger", - "greenscale": "greenscaleLogger", - "athina": "athinaLogger", - "traceloop": "traceloopLogger", - "s3": "s3Logger", - "openmeter": "openMeterLogger", -} +_SUCCESS_SINGLETONS: Final[Mapping[str, str]] = MappingProxyType( + { + "promptlayer": "promptLayerLogger", + "supabase": "supabaseClient", + "wandb": "weightsBiasesLogger", + "logfire": "logfireLogger", + "lunary": "lunaryLogger", + "helicone": "heliconeLogger", + "greenscale": "greenscaleLogger", + "athina": "athinaLogger", + "traceloop": "traceloopLogger", + "s3": "s3Logger", + "openmeter": "openMeterLogger", + } +) def dispatch_named_success( @@ -555,7 +588,9 @@ def dispatch_named_success( details: Final = view.model_call_details print_verbose: Final = _print_verbose() integration: Final = _legacy_module().get(_SUCCESS_SINGLETONS.get(name, "")) - without_response: Final = {key: value for key, value in details.items() if key != "original_response"} + without_response: Final = { # mutable-ok: legacy integrations receive a private mutable copy + key: value for key, value in details.items() if key != "original_response" + } match name: case "promptlayer" | "wandb" | "athina" if integration is not None: _method(integration, "log_event")( @@ -678,10 +713,12 @@ def _langfuse( view: Final = _logger(logger) module: Final = _legacy_module() - kwargs: Final = {key: value for key, value in view.model_call_details.items() if key != "original_response"} - select: Final = cast( + kwargs: Final = { # mutable-ok: langfuse receives a private mutable copy + key: value for key, value in view.model_call_details.items() if key != "original_response" + } + select: Final = cast( # cast-ok: legacy factory LegacyCall, langfuse_handler.LangFuseHandler.get_langfuse_logger_for_request - ) # cast-ok: legacy factory + ) handler: Final = select( globalLangfuseLogger=module.get("langFuseLogger"), standard_callback_dynamic_params=view.standard_callback_dynamic_params, @@ -689,7 +726,9 @@ def _langfuse( ) if handler is None: return - extra: Final[Details] = {"level": level, "status_message": status_message} if level is not None else {} + extra: Final[Mapping[str, object]] = ( + MappingProxyType({"level": level, "status_message": status_message}) if level is not None else _EMPTY + ) result: Final = _method(handler, "log_event_on_langfuse")( kwargs=kwargs, response_obj=response, @@ -699,7 +738,7 @@ def _langfuse( **extra, ) trace_id: Final = ( - cast(Details, result).get("trace_id") if isinstance(result, dict) else None + cast(Details, result).get("trace_id") if isinstance(result, dict) else None # cast-ok: legacy response dict ) # cast-ok: legacy response dict if trace_id is not None: _method(module["in_memory_trace_id_cache"], "set_cache")( @@ -719,6 +758,9 @@ def dispatch_named_failure( module: Final = _legacy_module() details: Final = view.model_call_details print_verbose: Final = _print_verbose() + without_response: Final = MappingProxyType( + {key: value for key, value in details.items() if key != "original_response"} + ) match name: case "lunary" if (lunary := module.get("lunaryLogger")) is not None: _method(lunary, "log_event")( @@ -771,8 +813,8 @@ def dispatch_named_failure( from litellm.integrations.logfire_logger import LogfireLevel _method(logfire, "log_event")( - kwargs={ - **{key: value for key, value in details.items() if key != "original_response"}, + kwargs={ # mutable-ok: logfire receives a private mutable copy + **without_response, "exception": exception, }, response_obj=None, @@ -805,11 +847,5 @@ def enqueue_background(coroutine: Awaitable[None]) -> None: from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER - enqueue: Final = cast( - Callable[[Awaitable[None]], None], GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue - ) # cast-ok: legacy worker accepts any coroutine + enqueue: Final = _call_of(_attribute(GLOBAL_LOGGING_WORKER, "ensure_initialized_and_enqueue")) contextvars.copy_context().run(enqueue, coroutine) - - -def now() -> Timestamp: - return datetime.datetime.now() diff --git a/litellm/rust_bridge/setup.py b/litellm/rust_bridge/setup.py index 820d39c7b3e..428f14ba480 100644 --- a/litellm/rust_bridge/setup.py +++ b/litellm/rust_bridge/setup.py @@ -2,12 +2,14 @@ from __future__ import annotations import datetime from collections.abc import Callable, Mapping, MutableSequence, Sequence -from typing import ( # noqa: TID251 # narrows legacy untyped registries at the boundary +from types import MappingProxyType +from typing import ( TYPE_CHECKING, Final, Literal, Protocol, - cast, + TypeAlias, + cast, # noqa: TID251 # narrows legacy untyped registries at the boundary ) from litellm.integrations.custom_logger import CustomLogger @@ -15,18 +17,24 @@ from litellm.integrations.custom_logger import CustomLogger if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging -CallbackTarget = str | Callable[..., object] | CustomLogger -RegistryName = Literal["input", "async_input", "success", "async_success", "failure", "async_failure", "callbacks"] +CallbackTarget: TypeAlias = str | Callable[..., object] | CustomLogger +RegistryName: TypeAlias = Literal[ + "input", "async_input", "success", "async_success", "failure", "async_failure", "callbacks" +] +NamedEvent: TypeAlias = Literal["success", "failure"] +Kwargs: TypeAlias = dict[str, object] # mutable-ok: the retained call kwargs are the dict callbacks receive and mutate -_REGISTRY_ATTRIBUTES: Final[Mapping[RegistryName, str]] = { - "input": "input_callback", - "async_input": "_async_input_callback", - "success": "success_callback", - "async_success": "_async_success_callback", - "failure": "failure_callback", - "async_failure": "_async_failure_callback", - "callbacks": "callbacks", -} +_REGISTRY_ATTRIBUTES: Final[Mapping[RegistryName, str]] = MappingProxyType( + { + "input": "input_callback", + "async_input": "_async_input_callback", + "success": "success_callback", + "async_success": "_async_success_callback", + "failure": "failure_callback", + "async_failure": "_async_failure_callback", + "callbacks": "callbacks", + } +) class _CallbackManager(Protocol): @@ -51,12 +59,12 @@ class _LoggingFactory(Protocol): function_id: str, call_type: str, start_time: datetime.datetime, - dynamic_success_callbacks: list[CallbackTarget] | None, - dynamic_failure_callbacks: list[CallbackTarget] | None, - dynamic_async_success_callbacks: list[CallbackTarget] | None, - dynamic_async_failure_callbacks: list[CallbackTarget] | None, - kwargs: dict[str, object], - applied_guardrails: list[str], + dynamic_success_callbacks: Sequence[CallbackTarget] | None, + dynamic_failure_callbacks: Sequence[CallbackTarget] | None, + dynamic_async_success_callbacks: Sequence[CallbackTarget] | None, + dynamic_async_failure_callbacks: Sequence[CallbackTarget] | None, + kwargs: Kwargs, + applied_guardrails: Sequence[str], supports_correlation_logging: bool, ) -> Logging: ... @@ -67,18 +75,30 @@ class _EnvironmentUpdater(Protocol): *, model: str | None, user: str, - optional_params: dict[str, object], - litellm_params: dict[str, object], + optional_params: Mapping[str, object], + litellm_params: Mapping[str, object], stream_options: object, ) -> None: ... -def registry(name: RegistryName) -> MutableSequence[CallbackTarget]: +def _legacy(value: object) -> Callable[..., object]: + return cast(Callable[..., object], value) # cast-ok: legacy module-level helpers are untyped + + +def _mapping(value: object) -> Mapping[str, object] | None: + if not isinstance(value, Mapping): + return None + return cast(Mapping[str, object], value) # cast-ok: isinstance narrows only to Mapping[Unknown, Unknown] + + +def registry( + name: RegistryName, +) -> MutableSequence[CallbackTarget]: # mutable-ok: the public litellm registries are mutated by contract import litellm - return cast( + return cast( # cast-ok: legacy module-level lists are untyped MutableSequence[CallbackTarget], getattr(litellm, _REGISTRY_ATTRIBUTES[name]) - ) # cast-ok: legacy module-level lists are untyped + ) def is_async_callable(callback: object) -> bool: @@ -97,11 +117,9 @@ def is_known_name(callback: str) -> bool: def resolve_named_integration(callback: str) -> CustomLogger | None: from litellm.litellm_core_utils import litellm_logging - resolve: Final = cast( # cast-ok: legacy factory is untyped at its definition - Callable[..., CustomLogger | None], - litellm_logging._init_custom_logger_compatible_class, # pyright: ignore[reportPrivateUsage] # legacy factory - ) - return resolve(callback, internal_usage_cache=None, llm_router=None) + factory: Final = _legacy(litellm_logging._init_custom_logger_compatible_class) # pyright: ignore[reportPrivateUsage, reportUnknownMemberType, reportUnknownArgumentType] # legacy factory + resolved: Final = factory(callback, internal_usage_cache=None, llm_router=None) + return resolved if isinstance(resolved, CustomLogger) else None def async_success_registry_has_type(callback: object) -> bool: @@ -118,15 +136,19 @@ def bootstrap(function_id: str | None) -> None: from litellm import utils from litellm.litellm_core_utils import cached_imports - combined: Final = list({*registry("input"), *registry("success"), *registry("failure")}) - utils.callback_list = cast( - list[str], combined - ) # rebind-ok: legacy module global consumed by set_callbacks # cast-ok: legacy list annotation is narrower than its contents - set_callbacks: Final = cast(Callable[..., None], cached_imports.get_set_callbacks()) # pyright: ignore[reportUnknownMemberType] # cast-ok: cached import is untyped + combined: Final = _bootstrap_list() + utils.callback_list = combined # rebind-ok: legacy module global consumed by set_callbacks + set_callbacks: Final = _legacy(cached_imports.get_set_callbacks()) # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # cached import is untyped set_callbacks(callback_list=combined, function_id=function_id) -def expand_named(callback: str, event: Literal["success", "failure"]) -> None: +def _bootstrap_list() -> list[str]: # mutable-ok: set_callbacks and the legacy module global expect a list + names: Final = frozenset({*registry("input"), *registry("success"), *registry("failure")}) + consumed: Final = cast(Sequence[str], names) # cast-ok: legacy list annotation is narrower than its contents + return list(consumed) # mutable-ok: consumed by set_callbacks + + +def expand_named(callback: str, event: NamedEvent) -> None: from litellm import utils utils._add_custom_logger_callback_to_specific_event(callback, event) # pyright: ignore[reportPrivateUsage] # legacy expansion helper @@ -168,25 +190,24 @@ def logger_fn(callback: object) -> None: def breadcrumb(kwargs: Mapping[str, object]) -> None: from litellm import utils - add_breadcrumb: Final = cast( - Callable[..., None] | None, utils.add_breadcrumb - ) # cast-ok: legacy sentry hook is untyped - if add_breadcrumb is None: + if utils.add_breadcrumb is None: return import litellm from litellm.litellm_core_utils import core_helpers - deep_copy: Final = cast( # cast-ok: legacy helper is untyped - Callable[[dict[str, object]], dict[str, object]], - core_helpers.safe_deep_copy, # pyright: ignore[reportUnknownMemberType] # legacy helper - ) - try: - copied: dict[str, object] = deep_copy(dict(kwargs)) - except Exception: # noqa: BLE001 # legacy breadcrumb falls back to the live mapping - copied = dict(kwargs) + deep_copy: Final = _legacy(core_helpers.safe_deep_copy) # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # legacy helper + copied: Final = _copied(deep_copy, kwargs) hidden: Final = frozenset(("messages", "input", "prompt")) if litellm.turn_off_message_logging else frozenset[str]() - details: Final = {key: value for key, value in copied.items() if key not in hidden} - add_breadcrumb(category="litellm.llm_call", message=f"Keyword Args: {details}", level="info") + details: Final = MappingProxyType({key: value for key, value in copied.items() if key not in hidden}) + _legacy(utils.add_breadcrumb)(category="litellm.llm_call", message=f"Keyword Args: {details}", level="info") + + +def _copied(deep_copy: Callable[..., object], kwargs: Mapping[str, object]) -> Mapping[str, object]: + try: + copied: Final = deep_copy(dict(kwargs)) # mutable-ok: safe_deep_copy expects a dict + except Exception: # noqa: BLE001 # legacy breadcrumb falls back to the live mapping + return kwargs + return _mapping(copied) or kwargs def prepare_environment() -> None: @@ -195,17 +216,38 @@ def prepare_environment() -> None: utils.custom_llm_setup() -def applied_guardrails(kwargs: Mapping[str, object]) -> list[str]: +def applied_guardrails(kwargs: Mapping[str, object]) -> Sequence[str]: from litellm.utils import get_applied_guardrails - return get_applied_guardrails(dict(kwargs)) + return tuple(get_applied_guardrails(dict(kwargs))) # mutable-ok: legacy helper expects a dict + + +def _litellm_params(kwargs: Mapping[str, object]) -> Mapping[str, object]: + metadata: Final = kwargs.get("metadata") + litellm_metadata: Final = kwargs.get("litellm_metadata") + base: Final = MappingProxyType({"api_base": ""}) + with_metadata: Final = MappingProxyType({**base, "metadata": metadata}) if "metadata" in kwargs else base + typed_metadata: Final = _mapping(litellm_metadata) + if typed_metadata is None: + return with_metadata + with_litellm_metadata: Final = MappingProxyType({**with_metadata, "litellm_metadata": typed_metadata}) + if metadata: + return with_litellm_metadata + copied_metadata: Final = dict(typed_metadata) # mutable-ok: callbacks may mutate this copy + return MappingProxyType({**with_litellm_metadata, "metadata": copied_metadata}) + + +def _owned( + callbacks: Sequence[CallbackTarget] | None, +) -> list[CallbackTarget] | None: # mutable-ok: Logging appends to these lists + return list(callbacks) if callbacks is not None else None # mutable-ok: Logging appends to these lists def build_logging( *, call_type: str, model: str | None, - kwargs: dict[str, object], + kwargs: Kwargs, start_time: datetime.datetime, asynchronous: bool, dynamic_success: Sequence[CallbackTarget] | None, @@ -216,7 +258,6 @@ def build_logging( from litellm.litellm_core_utils.cached_imports import get_litellm_logging_class function_id: Final = kwargs.get("id") - metadata: Final = kwargs.get("metadata") trace_id: Final = kwargs.get("litellm_trace_id") factory: Final = cast(_LoggingFactory, get_litellm_logging_class()) # cast-ok: legacy constructor is untyped logger: Final = factory( @@ -228,35 +269,20 @@ def build_logging( function_id=function_id if isinstance(function_id, str) else "", call_type=call_type, start_time=start_time, - dynamic_success_callbacks=list(dynamic_success) if dynamic_success is not None else None, - dynamic_failure_callbacks=list(dynamic_failure) if dynamic_failure is not None else None, - dynamic_async_success_callbacks=list(dynamic_async_success) if dynamic_async_success is not None else None, + dynamic_success_callbacks=_owned(dynamic_success), + dynamic_failure_callbacks=_owned(dynamic_failure), + dynamic_async_success_callbacks=_owned(dynamic_async_success), dynamic_async_failure_callbacks=None, kwargs=kwargs, - applied_guardrails=list(guardrails), + applied_guardrails=list(guardrails), # mutable-ok: legacy constructor annotation is list supports_correlation_logging=asynchronous, ) - litellm_metadata: Final = kwargs.get("litellm_metadata") - litellm_params: Final[dict[str, object]] = { - "api_base": "", - **({"metadata": kwargs["metadata"]} if "metadata" in kwargs else {}), - **( - { - "litellm_metadata": litellm_metadata, - **( - {} if metadata else {"metadata": dict(cast(Mapping[str, object], litellm_metadata))} - ), # cast-ok: isinstance narrows only to dict[Unknown, Unknown] - } - if isinstance(litellm_metadata, dict) - else {} - ), - } update: Final = cast(_EnvironmentUpdater, logger.update_environment_variables) # cast-ok: legacy method is untyped update( model=model, user="", - optional_params={}, - litellm_params=litellm_params, + optional_params=MappingProxyType({}), + litellm_params=_litellm_params(kwargs), stream_options=kwargs.get("stream_options"), ) return logger diff --git a/tests/test_litellm/rust_bridge/test_bindings.py b/tests/test_litellm/rust_bridge/test_bindings.py index 7e6fea7adfb..34e5e2c04ae 100644 --- a/tests/test_litellm/rust_bridge/test_bindings.py +++ b/tests/test_litellm/rust_bridge/test_bindings.py @@ -46,8 +46,4 @@ def test_decline_accessor_only_catches_admission_declines(monkeypatch: pytest.Mo native: Final = SimpleNamespace(RustBridgeDeclined=Declined, RustUpstreamError=Upstream) if available else None monkeypatch.setattr(bindings, "get_native_bridge", lambda: native) assert bindings.native_decline_types() == ((Declined,) if available else ()) - with pytest.raises(Upstream): - try: - raise Upstream("already dispatched") - except bindings.native_decline_types(): - pytest.fail("upstream failure was allowed to replay") + assert not isinstance(Upstream("already dispatched"), bindings.native_decline_types())