From 58be89ffbae946a258858a455eb2589f00abf047 Mon Sep 17 00:00:00 2001 From: Yujong Lee Date: Wed, 16 Sep 2026 19:07:06 -0700 Subject: [PATCH] Revert "style(rust_bridge): satisfy strict lint gates in the callback leaves" This reverts commit aecc08050f1c39683beee10ec9fc2272cf7c7025. --- litellm/ocr/main.py | 1 - litellm/rust_bridge/leaves.py | 202 +++++++----------- litellm/rust_bridge/setup.py | 172 +++++++-------- .../test_litellm/rust_bridge/test_bindings.py | 6 +- 4 files changed, 161 insertions(+), 220 deletions(-) diff --git a/litellm/ocr/main.py b/litellm/ocr/main.py index c6fca6078e5..1839ab95526 100644 --- a/litellm/ocr/main.py +++ b/litellm/ocr/main.py @@ -39,7 +39,6 @@ async def aocr(*args: object, **kwargs: object) -> OCRResponse: # kwargs-ok: pr ) return await fallback(*args, **kwargs) - def _decline_types() -> tuple[type[BaseException], ...]: exception_types: Final = native_exception_types() return (exception_types[0],) if exception_types is not None else () diff --git a/litellm/rust_bridge/leaves.py b/litellm/rust_bridge/leaves.py index 6b9020aba78..64e239d1615 100644 --- a/litellm/rust_bridge/leaves.py +++ b/litellm/rust_bridge/leaves.py @@ -17,14 +17,12 @@ import datetime import json import traceback from collections.abc import Awaitable, Callable, Mapping -from types import MappingProxyType -from typing import ( +from typing import ( # noqa: TID251 # narrows the untyped legacy Logging and CustomLogger surfaces once TYPE_CHECKING, Final, Literal, Protocol, - TypeAlias, - cast, # noqa: TID251 # narrows the untyped legacy Logging and CustomLogger surfaces once + cast, ) from litellm.integrations.custom_logger import CustomLogger @@ -32,14 +30,12 @@ from litellm.integrations.custom_logger import CustomLogger if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging -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]] +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]] class LoggerView(Protocol): @@ -64,19 +60,17 @@ 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: Mapping[str, str], additional_args: Details - ) -> None: ... + def _print_llm_call_debugging_log(self, api_base: str, headers: Details, additional_args: Details) -> None: ... def _get_request_curl_command( - self, api_base: str, headers: Mapping[str, str] | None, additional_args: Details, data: object + self, api_base: str, headers: Details | 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: Mapping[str, str]) -> Details: ... + def _get_masked_headers(self, headers: Details) -> Details: ... def _response_cost_calculator(self, result: object) -> float | None: ... @@ -143,31 +137,6 @@ 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 @@ -179,9 +148,9 @@ def _integration(callback: CustomLogger) -> IntegrationView: def _legacy_module() -> Mapping[str, object]: from litellm.litellm_core_utils import litellm_logging - return cast( # cast-ok: module globals hold the legacy integration singletons + return cast( Mapping[str, object], vars(litellm_logging) - ) + ) # cast-ok: module globals hold the legacy integration singletons def _print_verbose() -> LegacyCall: @@ -191,18 +160,17 @@ def _print_verbose() -> LegacyCall: def _method(target: object, name: str) -> LegacyCall: - return _call_of(_attribute(target, name)) + return cast(LegacyCall, getattr(target, name)) # cast-ok: legacy integration singletons are untyped def _async_method(target: object, name: str) -> LegacyAsyncCall: - return cast(LegacyAsyncCall, _attribute(target, name)) # cast-ok: legacy integration singletons are untyped + return cast(LegacyAsyncCall, getattr(target, name)) # cast-ok: legacy integration singletons are untyped def _redact_string(value: str) -> str: from litellm.litellm_core_utils import litellm_logging - redact: Final = _call_of(litellm_logging._redact_string) # pyright: ignore[reportPrivateUsage] # private legacy helper - return str(redact(value)) + return cast(Callable[[str], str], litellm_logging._redact_string)(value) # pyright: ignore[reportPrivateUsage] # cast-ok: legacy helper def _redact_result(details: Details, result: object) -> object: @@ -217,17 +185,13 @@ def record_pre_call( *, api_key: str | None, body: Details, - headers: Mapping[str, str], + headers: dict[str, str], url: str, ) -> None: view: Final = _logger(logger) - additional_args: Final[Details] = { # mutable-ok: legacy additional_args is rebound by callbacks in place - "complete_input_dict": body, - "headers": headers, - "api_base": url, - } + additional_args: Final[Details] = {"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=headers, additional_args=additional_args) # pyright: ignore[reportPrivateUsage] # legacy debug output + view._print_llm_call_debugging_log(api_base=url, headers=dict(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() @@ -240,13 +204,15 @@ 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 - metadata: Final = _seed(_seed(details, "litellm_params"), "metadata") + 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) 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(Mapping[str, str], additional_args.get("headers") or _EMPTY) # cast-ok: legacy nested dict - body: Final = additional_args.get("complete_input_dict", _EMPTY) + headers: Final = cast(Details, additional_args.get("headers") or {}) # cast-ok: legacy nested dict + body: Final = additional_args.get("complete_input_dict", {}) 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 @@ -266,16 +232,18 @@ 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 = getattr(logger, "logger_fn", None) + logger_fn: Final = cast( + Callable[[Details], object] | None, getattr(logger, "logger_fn", None) + ) # cast-ok: user hook is untyped if not callable(logger_fn): return try: - _call_of(logger_fn)(_logger(logger).model_call_details) + 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: Mapping[str, str]) -> None: +def record_post_call(logger: Logging, *, original_response: object, body: Details, headers: dict[str, str]) -> None: view: Final = _logger(logger) serialized: Final = ( json.dumps(original_response, default=str) if isinstance(original_response, dict) else original_response @@ -284,7 +252,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}, # mutable-ok: rebound in place by callbacks + additional_args={"complete_input_dict": body, "headers": headers}, ) _run_logger_fn(logger) _redact_result(view.model_call_details, serialized) @@ -362,9 +330,12 @@ 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 = _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 + 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 existing: Final = details.get("response_cost") if "response_cost" in hidden: details["response_cost"] = hidden["response_cost"] @@ -374,7 +345,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-ok: legacy builder returns the payload TypedDict + cast(StandardLoggingPayload, payload) ) # cast-ok: legacy builder returns the payload TypedDict return _redact_result(details, response) @@ -397,16 +368,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, 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 + 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( 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=_EMPTY, + init_response_obj={}, start_time=start_time, end_time=end_time, logging_obj=logger, @@ -418,19 +389,17 @@ def prepare_failure_logging( return formatted -_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", - } -) +_EVENT_HOOKS: Final[Mapping[TerminalFamily, str]] = { + "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 = _details_of(view.model_call_details.get("litellm_params") or _EMPTY) + params: Final = cast(Details, view.model_call_details.get("litellm_params") or {}) # cast-ok: legacy nested dict return view.should_run_callback(callback=callback, litellm_params=params, event_hook=_EVENT_HOOKS[family]) @@ -458,9 +427,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( # cast-ok: legacy helper + redact: Final = cast( LegacyCall, redact_messages.redact_message_input_output_from_custom_logger - ) + ) # cast-ok: legacy helper redacted: Final = ( result if isinstance(callback, CustomGuardrail) @@ -498,9 +467,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( # cast-ok: legacy helper + redact: Final = cast( 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 @@ -564,21 +533,19 @@ def dispatch_callable( ) -_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", - } -) +_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", +} def dispatch_named_success( @@ -588,9 +555,7 @@ 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 = { # mutable-ok: legacy integrations receive a private mutable copy - key: value for key, value in details.items() if key != "original_response" - } + without_response: Final = {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")( @@ -713,12 +678,10 @@ def _langfuse( view: Final = _logger(logger) module: Final = _legacy_module() - 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 + kwargs: Final = {key: value for key, value in view.model_call_details.items() if key != "original_response"} + select: Final = cast( 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, @@ -726,9 +689,7 @@ def _langfuse( ) if handler is None: return - extra: Final[Mapping[str, object]] = ( - MappingProxyType({"level": level, "status_message": status_message}) if level is not None else _EMPTY - ) + extra: Final[Details] = {"level": level, "status_message": status_message} if level is not None else {} result: Final = _method(handler, "log_event_on_langfuse")( kwargs=kwargs, response_obj=response, @@ -738,7 +699,7 @@ def _langfuse( **extra, ) trace_id: Final = ( - cast(Details, result).get("trace_id") if isinstance(result, dict) else None # cast-ok: legacy response dict + cast(Details, result).get("trace_id") if isinstance(result, dict) else None ) # cast-ok: legacy response dict if trace_id is not None: _method(module["in_memory_trace_id_cache"], "set_cache")( @@ -758,9 +719,6 @@ 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")( @@ -813,8 +771,8 @@ def dispatch_named_failure( from litellm.integrations.logfire_logger import LogfireLevel _method(logfire, "log_event")( - kwargs={ # mutable-ok: logfire receives a private mutable copy - **without_response, + kwargs={ + **{key: value for key, value in details.items() if key != "original_response"}, "exception": exception, }, response_obj=None, @@ -847,5 +805,11 @@ def enqueue_background(coroutine: Awaitable[None]) -> None: from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER - enqueue: Final = _call_of(_attribute(GLOBAL_LOGGING_WORKER, "ensure_initialized_and_enqueue")) + enqueue: Final = cast( + Callable[[Awaitable[None]], None], GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue + ) # cast-ok: legacy worker accepts any coroutine 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 428f14ba480..820d39c7b3e 100644 --- a/litellm/rust_bridge/setup.py +++ b/litellm/rust_bridge/setup.py @@ -2,14 +2,12 @@ from __future__ import annotations import datetime from collections.abc import Callable, Mapping, MutableSequence, Sequence -from types import MappingProxyType -from typing import ( +from typing import ( # noqa: TID251 # narrows legacy untyped registries at the boundary TYPE_CHECKING, Final, Literal, Protocol, - TypeAlias, - cast, # noqa: TID251 # narrows legacy untyped registries at the boundary + cast, ) from litellm.integrations.custom_logger import CustomLogger @@ -17,24 +15,18 @@ from litellm.integrations.custom_logger import CustomLogger if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import Logging -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 +CallbackTarget = str | Callable[..., object] | CustomLogger +RegistryName = Literal["input", "async_input", "success", "async_success", "failure", "async_failure", "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", - } -) +_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", +} class _CallbackManager(Protocol): @@ -59,12 +51,12 @@ class _LoggingFactory(Protocol): function_id: str, call_type: str, start_time: datetime.datetime, - 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], + 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], supports_correlation_logging: bool, ) -> Logging: ... @@ -75,30 +67,18 @@ class _EnvironmentUpdater(Protocol): *, model: str | None, user: str, - optional_params: Mapping[str, object], - litellm_params: Mapping[str, object], + optional_params: dict[str, object], + litellm_params: dict[str, object], stream_options: object, ) -> None: ... -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 +def registry(name: RegistryName) -> MutableSequence[CallbackTarget]: import litellm - return cast( # cast-ok: legacy module-level lists are untyped + return cast( MutableSequence[CallbackTarget], getattr(litellm, _REGISTRY_ATTRIBUTES[name]) - ) + ) # cast-ok: legacy module-level lists are untyped def is_async_callable(callback: object) -> bool: @@ -117,9 +97,11 @@ def is_known_name(callback: str) -> bool: def resolve_named_integration(callback: str) -> CustomLogger | None: from litellm.litellm_core_utils import litellm_logging - 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 + 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) def async_success_registry_has_type(callback: object) -> bool: @@ -136,19 +118,15 @@ def bootstrap(function_id: str | None) -> None: from litellm import utils from litellm.litellm_core_utils import cached_imports - 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 + 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 set_callbacks(callback_list=combined, function_id=function_id) -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: +def expand_named(callback: str, event: Literal["success", "failure"]) -> None: from litellm import utils utils._add_custom_logger_callback_to_specific_event(callback, event) # pyright: ignore[reportPrivateUsage] # legacy expansion helper @@ -190,24 +168,25 @@ def logger_fn(callback: object) -> None: def breadcrumb(kwargs: Mapping[str, object]) -> None: from litellm import utils - if utils.add_breadcrumb is None: + add_breadcrumb: Final = cast( + Callable[..., None] | None, utils.add_breadcrumb + ) # cast-ok: legacy sentry hook is untyped + if add_breadcrumb is None: return import litellm from litellm.litellm_core_utils import core_helpers - 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 = 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]: + 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: Final = deep_copy(dict(kwargs)) # mutable-ok: safe_deep_copy expects a dict + copied: dict[str, object] = deep_copy(dict(kwargs)) except Exception: # noqa: BLE001 # legacy breadcrumb falls back to the live mapping - return kwargs - return _mapping(copied) or kwargs + copied = dict(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") def prepare_environment() -> None: @@ -216,38 +195,17 @@ def prepare_environment() -> None: utils.custom_llm_setup() -def applied_guardrails(kwargs: Mapping[str, object]) -> Sequence[str]: +def applied_guardrails(kwargs: Mapping[str, object]) -> list[str]: from litellm.utils import get_applied_guardrails - 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 + return get_applied_guardrails(dict(kwargs)) def build_logging( *, call_type: str, model: str | None, - kwargs: Kwargs, + kwargs: dict[str, object], start_time: datetime.datetime, asynchronous: bool, dynamic_success: Sequence[CallbackTarget] | None, @@ -258,6 +216,7 @@ 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( @@ -269,20 +228,35 @@ def build_logging( function_id=function_id if isinstance(function_id, str) else "", call_type=call_type, start_time=start_time, - dynamic_success_callbacks=_owned(dynamic_success), - dynamic_failure_callbacks=_owned(dynamic_failure), - dynamic_async_success_callbacks=_owned(dynamic_async_success), + 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_async_failure_callbacks=None, kwargs=kwargs, - applied_guardrails=list(guardrails), # mutable-ok: legacy constructor annotation is list + applied_guardrails=list(guardrails), 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=MappingProxyType({}), - litellm_params=_litellm_params(kwargs), + optional_params={}, + litellm_params=litellm_params, 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 34e5e2c04ae..7e6fea7adfb 100644 --- a/tests/test_litellm/rust_bridge/test_bindings.py +++ b/tests/test_litellm/rust_bridge/test_bindings.py @@ -46,4 +46,8 @@ 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 ()) - assert not isinstance(Upstream("already dispatched"), bindings.native_decline_types()) + with pytest.raises(Upstream): + try: + raise Upstream("already dispatched") + except bindings.native_decline_types(): + pytest.fail("upstream failure was allowed to replay")