mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-19 00:01:29 +00:00
style(rust_bridge): satisfy strict lint gates in the callback leaves
Type aliases carry TypeAlias, cast sites carry cast-ok on the call line, dict fallbacks use frozen constants, and nested envelope seeding replaces a None metadata value instead of returning it (legacy behaviour the setdefault form missed). Fixes the PT012 shape in test_bindings and formats ocr/main.
This commit is contained in:
Yujong Lee
parent
316700aeb8
commit
aecc08050f
4 changed files with 220 additions and 163 deletions
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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())
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue