scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-15 23:31:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 165

just pass lint

Browse source
This commit is contained in:
Yujong Lee 2026-09-07 15:15:31 -07:00
parent 5afda69ca5
commit f250e3dded
4 changed files with 70 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
litellm/ocr/main.py
View file

@ -180,7 +180,7 @@ async def aocr(
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
extra_headers: dict[str, object] | None = None,
**kwargs: object,
**kwargs: object, # kwargs-ok: public SDK forwards provider-specific OCR parameters
) -> OCRResponse:
"""
Async OCR function.
@ -251,7 +251,7 @@ async def _legacy_aocr(
timeout: float | httpx.Timeout | None = None,
custom_llm_provider: str | None = None,
extra_headers: dict[str, object] | None = None,
**kwargs: object,
**kwargs: object, # kwargs-ok: public SDK forwards provider-specific OCR parameters
) -> OCRResponse:
completion_kwargs: Final[dict[str, object]] = {
"model": model,

90
litellm/rust_bridge/ocr.py
View file

@ -66,37 +66,69 @@ def initialize_logging(arguments: dict[str, object], asynchronous: bool) -> obje
supplied: Final = arguments.get("litellm_logging_obj")
if supplied is not None:
return supplied
callbacks: Final = tuple(dict.fromkeys(utils.get_dynamic_callbacks(cast(list, arguments.get("callbacks")))))
success: Final = tuple(dict.fromkeys((*callbacks, *cast(list, arguments.get("success_callback") or []))))
failure: Final = tuple(dict.fromkeys((*callbacks, *cast(list, arguments.get("failure_callback") or []))))
callbacks: Final = tuple( # cast-ok: callback registry accepts heterogeneous legacy callback objects
dict.fromkeys(
utils.get_dynamic_callbacks(
cast( # cast-ok: callback registry accepts heterogeneous legacy callback objects
list, arguments.get("callbacks")
) # cast-ok: callback registry accepts heterogeneous legacy callback objects
) # cast-ok: callback registry accepts heterogeneous legacy callback objects
) # cast-ok: callback registry accepts heterogeneous legacy callback objects # mutable-ok: deduplication uses dict keys
)
success: Final = tuple( # cast-ok: per-call callback list is a legacy untyped boundary
dict.fromkeys(
(
*callbacks,
*cast( # cast-ok: per-call callback list is a legacy untyped boundary
list, arguments.get("success_callback") or []
), # cast-ok: per-call callback list is a legacy untyped boundary
) # cast-ok: per-call callback list is a legacy untyped boundary
) # cast-ok: per-call callback list is a legacy untyped boundary # mutable-ok: deduplication uses dict keys
)
failure: Final = tuple( # cast-ok: per-call callback list is a legacy untyped boundary
dict.fromkeys(
(
*callbacks,
*cast( # cast-ok: per-call callback list is a legacy untyped boundary
list, arguments.get("failure_callback") or []
), # cast-ok: per-call callback list is a legacy untyped boundary
) # cast-ok: per-call callback list is a legacy untyped boundary
) # cast-ok: per-call callback list is a legacy untyped boundary # mutable-ok: deduplication uses dict keys
)
configured: Final = tuple(
dict.fromkeys(
(
*litellm.input_callback,
*litellm.success_callback,
*litellm.failure_callback,
*litellm._async_success_callback,
*litellm._async_failure_callback,
*litellm._async_success_callback, # pyright: ignore[reportPrivateUsage] # callback registry has no public accessor
*litellm._async_failure_callback, # pyright: ignore[reportPrivateUsage] # callback registry has no public accessor
*success,
*failure,
)
)
)
uninitialized: Final = [
uninitialized: Final = [ # mutable-ok: set_callbacks requires a mutable callback list
cb
for cb in configured
if isinstance(cb, str)
and (
cb not in litellm._known_custom_logger_compatible_callbacks
cb not in litellm._known_custom_logger_compatible_callbacks # pyright: ignore[reportPrivateUsage] # callback compatibility registry has no public accessor
or cb in litellm.input_callback + litellm.success_callback + litellm.failure_callback
)
and cb not in (utils.callback_list or [])
and cb
not in (utils.callback_list or []) # mutable-ok: empty list normalizes an uninitialized callback registry
]
if uninitialized:
set_callbacks(uninitialized, function_id=arguments.get("id"))
utils.callback_list = list(dict.fromkeys((*(utils.callback_list or []), *uninitialized)))
if litellm_logging.customLogger is None:
set_callbacks([cb for cb in configured if callable(cb)], function_id=arguments.get("id"))
utils.callback_list = list( # mutable-ok: global callback registry is mutable
dict.fromkeys((*(utils.callback_list or []), *uninitialized))
) # mutable-ok: global callback registry is mutable
if litellm_logging.customLogger is None: # pyright: ignore[reportUnnecessaryComparison] # runtime plugin registry can be reset to None
set_callbacks(
[cb for cb in configured if callable(cb)], # mutable-ok: set_callbacks requires a mutable callback list
function_id=arguments.get("id"), # mutable-ok: set_callbacks requires a mutable callback list
) # mutable-ok: set_callbacks requires a mutable callback list
for event, registered, add_async in (
("input", litellm.input_callback, litellm.logging_callback_manager.add_litellm_input_callback),
("success", litellm.success_callback, litellm.logging_callback_manager.add_litellm_async_success_callback),
@ -107,11 +139,11 @@ def initialize_logging(arguments: dict[str, object], asynchronous: bool) -> obje
if cb not in getattr(litellm, f"_async_{event}_callback"):
add_async(cb)
registered.remove(cb)
elif event != "input" and isinstance(cb, str) and cb in litellm._known_custom_logger_compatible_callbacks:
utils._add_custom_logger_callback_to_specific_event(cb, event)
elif event != "input" and isinstance(cb, str) and cb in litellm._known_custom_logger_compatible_callbacks: # pyright: ignore[reportPrivateUsage] # callback compatibility registry has no public accessor
utils._add_custom_logger_callback_to_specific_event(cb, event) # pyright: ignore[reportPrivateUsage] # callback manager only exposes this internal registration path
for event, registered, add_sync in (
("success", litellm._async_success_callback, litellm.logging_callback_manager.add_litellm_success_callback),
("failure", litellm._async_failure_callback, litellm.logging_callback_manager.add_litellm_failure_callback),
("success", litellm._async_success_callback, litellm.logging_callback_manager.add_litellm_success_callback), # pyright: ignore[reportPrivateUsage] # callback registry has no public accessor
("failure", litellm._async_failure_callback, litellm.logging_callback_manager.add_litellm_failure_callback), # pyright: ignore[reportPrivateUsage] # callback registry has no public accessor
):
for cb in tuple(registered):
if callable(cb) and not isinstance(cb, CustomLogger) and not coroutine_checker.is_async_callable(cb):
@ -124,29 +156,33 @@ def initialize_logging(arguments: dict[str, object], asynchronous: bool) -> obje
messages="default-message-value",
stream=False,
call_type="aocr" if asynchronous else "ocr",
start_time=datetime.now(),
start_time=datetime.now(), # noqa: DTZ005 # Logging preserves the legacy naive timestamp contract
litellm_call_id=call_id,
function_id=str(arguments.get("id") or ""),
litellm_trace_id=cast(str | None, arguments.get("litellm_trace_id")),
dynamic_input_callbacks=[
litellm_trace_id=cast( # cast-ok: public call argument is validated by Logging
str | None, arguments.get("litellm_trace_id")
), # cast-ok: public call argument is validated by Logging
dynamic_input_callbacks=[ # mutable-ok: Logging callback configuration is mutable
cb for cb in callbacks if cb not in litellm.input_callback and not coroutine_checker.is_async_callable(cb)
],
dynamic_success_callbacks=[
dynamic_success_callbacks=[ # mutable-ok: Logging callback configuration is mutable
cb for cb in success if not coroutine_checker.is_async_callable(cb) and cb not in ("dynamodb", "s3")
],
dynamic_async_success_callbacks=[
dynamic_async_success_callbacks=[ # mutable-ok: Logging callback configuration is mutable
cb
for cb in success
if coroutine_checker.is_async_callable(cb) or isinstance(cb, CustomLogger) or cb in ("dynamodb", "s3")
],
dynamic_failure_callbacks=[cb for cb in failure if not coroutine_checker.is_async_callable(cb)],
dynamic_async_failure_callbacks=[
dynamic_failure_callbacks=[ # mutable-ok: Logging callback configuration is mutable
cb for cb in failure if not coroutine_checker.is_async_callable(cb)
], # mutable-ok: Logging callback configuration is mutable
dynamic_async_failure_callbacks=[ # mutable-ok: Logging callback configuration is mutable
cb for cb in failure if coroutine_checker.is_async_callable(cb) or isinstance(cb, CustomLogger)
],
kwargs=arguments,
supports_correlation_logging=asynchronous,
)
logger.dynamic_input_callbacks = [
logger.dynamic_input_callbacks = [ # mutable-ok: remove callbacks promoted to the global registry
cb for cb in dict.fromkeys(logger.dynamic_input_callbacks or []) if cb not in litellm.input_callback
]
arguments["litellm_call_id"] = call_id
@ -161,7 +197,7 @@ def invoke_terminal(
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
logging: Final = cast(Logging, logger)
logging: Final = cast(Logging, logger) # cast-ok: Rust passes the logger returned by initialize_logging
if action == "sync_success":
def run() -> None:
@ -179,15 +215,15 @@ def invoke_terminal(
GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(async_coroutine=run_async())
if getattr(logging, "_defer_async_logging", False) is True:
logging._enqueue_deferred_logging = enqueue
logging._enqueue_deferred_logging = enqueue # pyright: ignore[reportPrivateUsage] # preserves Logging's deferred callback contract
else:
enqueue()
return None
if action == "sync_success_if_needed":
if logging._should_run_sync_callbacks_for_async_calls():
if logging._should_run_sync_callbacks_for_async_calls(): # pyright: ignore[reportPrivateUsage] # preserves Logging's async callback policy
return invoke_terminal("sync_success", roots, logger, value, start_time, end_time)
return None
exception: Final = cast(Exception, value)
exception: Final = cast(Exception, value) # cast-ok: Rust routes terminal failure values as Python exceptions
trace: Final = "".join(traceback.format_exception(type(exception), exception, exception.__traceback__))
if action == "sync_failure":
logging.failure_handler(exception, trace, start_time, end_time)

8
tests/test_litellm/ocr/test_rust_bridge.py
View file

@ -705,7 +705,7 @@ async def test_private_native_callback_lifecycle(
)
try:
if failure:
with pytest.raises(litellm.RateLimitError) as caught:
with pytest.raises(litellm.RateLimitError) as caught: # noqa: PT012 # parametrized sync and async calls require distinct statements
if asynchronous:
await litellm.aocr(**arguments)
else:
@ -839,7 +839,7 @@ async def test_private_native_callable_terminal_callback(
}
try:
if failure:
with pytest.raises(litellm.RateLimitError) as caught:
with pytest.raises(litellm.RateLimitError) as caught: # noqa: PT012 # parametrized sync and async calls require distinct statements
if asynchronous:
await litellm.aocr(**arguments)
else:
@ -983,7 +983,7 @@ async def test_native_unsupported_requests_never_prepare_or_send(
"num_retries": 0,
**options,
}
with pytest.raises(NotImplementedError, match=message):
with pytest.raises(NotImplementedError, match=message): # noqa: PT012 # parametrized native and public entry points differ
function = litellm.aocr if asynchronous else litellm.ocr
result = (function if public else inspect.unwrap(function))(**arguments)
if asynchronous:
@ -1127,7 +1127,7 @@ async def test_native_callback_escape_never_sends_or_replays(
monkeypatch.setattr(litellm, "input_callback", [Abort()])
arguments = dict(model=MODEL, document=document, api_key="sk-test", api_base=wire_recorder.api_base, num_retries=0)
with pytest.raises(PreCallAbort) as caught:
with pytest.raises(PreCallAbort) as caught: # noqa: PT012 # parametrized sync and async calls require distinct statements
if asynchronous:
await litellm.aocr(**arguments)
else:

2
tests/test_litellm/rust_bridge/native_route_wheel_test.py
View file

@ -367,7 +367,7 @@ async def exercise_unsupported_ocr(native: object, api_base: str) -> None:
else:
native.ocr(arguments)
except NotImplementedError as error:
assert operation in str(error)
assert operation in str(error) # noqa: PT017 # the selected native entry point is parametrized at runtime
else:
raise AssertionError(f"native OCR accepted unsupported operation: {operation}")
assert arguments["litellm_logging_obj"].calls == ()