scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-14 04:09:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 97

chore(batches): keep the lint gates within main's ceilings after the merge

Browse source 
Typed the helpers main now types (Mapping litellm_params, Any Span), replaced the
banned cast with a literal match, dropped isinstance checks main made redundant,
and moved test-quality reasons onto the lines the gate reads.
This commit is contained in:
yucheng-berri 2026-10-08 15:08:46 -07:00
parent c653e2e771
commit e00471b86a
11 changed files with 51 additions and 80 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py
View file

@ -940,7 +940,7 @@ class CheckBatchCost:
batch_failed_requests=batch_result.failed_requests,
batch_prompt_cost=batch_result.prompt_cost,
batch_completion_cost=batch_result.completion_cost,
batch_output_file_content=content_bytes if isinstance(content_bytes, bytes) else None,
batch_output_file_content=content_bytes,
batch_error_file_content=error_file_content,
)
except Exception:

14
litellm/batches/batch_line_item_logging.py
View file

@ -3,7 +3,7 @@ import uuid
from collections.abc import Iterator, Mapping
from datetime import datetime
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, Literal, TypeAlias, cast, get_args
from typing import TYPE_CHECKING, Final, Literal, TypeAlias
from typing_extensions import assert_never
@ -35,7 +35,6 @@ _BatchLineProvider: TypeAlias = Literal[
"openai", "azure", "vertex_ai", "hosted_vllm", "anthropic", "bedrock", "mistral"
]
_SUPPORTED_LINE_PROVIDERS: Final = frozenset(get_args(_BatchLineProvider))
_SECRET_PARAM_KEYS: Final = frozenset(
{
@ -53,9 +52,11 @@ _SECRET_PARAM_KEYS: Final = frozenset(
def _supported_line_provider(value: str) -> _BatchLineProvider | None:
if value in _SUPPORTED_LINE_PROVIDERS:
return cast("_BatchLineProvider", value) # cast-ok: membership in the literal's args was just checked
return None
match value:
case "openai" | "azure" | "vertex_ai" | "hosted_vllm" | "anthropic" | "bedrock" | "mistral":
return value
case _:
return None
_CALL_TYPE_BY_BATCH_URL: Final = MappingProxyType(
@ -332,8 +333,7 @@ async def _emit_line_event(
status_code: Final = _line_status_code(entry, custom_llm_provider)
request_call_type: Final = _call_type_for_request(request_line)
response_body: Final = _get_response_from_batch_job_output_file(entry, custom_llm_provider)
parent_start_time: Final = parent.start_time # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # Logging.start_time is untyped upstream
start_time: Final = parent_start_time if isinstance(parent_start_time, datetime) else datetime.now() # noqa: DTZ005 # naive to match the logging pipeline start_time
start_time: Final = parent.start_time
parent_params: Final = _as_object_mapping(parent.litellm_params) or _EMPTY_BODY # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # Logging.litellm_params is untyped upstream
model: Final = _line_model(response_body, request_body, parent)

4
litellm/batches/batch_utils.py
View file

@ -139,7 +139,7 @@ async def handle_completed_batch_with_files(
batch: Batch,
custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic", "bedrock", "mistral"],
model_name: str | None = None,
litellm_params: dict[str, object] | None = None, # mutable-ok: file fetchers require a plain dict
litellm_params: Mapping[str, object] | None = None,
model_info: ModelInfo | None = None,
) -> "tuple[BatchCostUsageResult, BatchResultFiles]":
"""_handle_completed_batch plus the raw output/error bytes it fetched, so
@ -522,7 +522,7 @@ async def _fetch_batch_output_file_content(
async def fetch_batch_error_file_content(
batch: Batch,
custom_llm_provider: Literal["openai", "azure", "vertex_ai", "hosted_vllm", "anthropic", "bedrock", "mistral"],
litellm_params: dict[str, object] | None, # mutable-ok: file fetchers require a plain dict
litellm_params: Mapping[str, object] | None,
) -> bytes | None:
"""Fetch the batch's separate error file bytes; None when it has none or the fetch fails."""
if batch.error_file_id is None:

24
litellm/litellm_core_utils/core_helpers.py
View file

@ -19,9 +19,9 @@ if TYPE_CHECKING:
from litellm.types.utils import ModelResponseStream
Span = _Span | object
Span = _Span | Any
else:
Span = object
Span = Any
_CODEX_CLIENT_PREFIX_RE: Final = re.compile(r"^codex[-_ /]", re.IGNORECASE)
@ -271,7 +271,7 @@ def remove_index_from_tool_calls(
tool_call.pop("index", None)
def remove_items_at_indices(items: list[object] | None, indices: Iterable[int]) -> None:
def remove_items_at_indices(items: list[Any] | None, indices: Iterable[int]) -> None:
"""Remove items from a list in-place by index"""
if items is None:
return
@ -728,9 +728,9 @@ def filter_internal_params(data: dict, additional_internal_params: set | None =
def redact_nested_match_and_regex_keys(
payload: dict | list[object] | str | None,
payload: dict | list[Any] | str | None,
keys: Collection[str] = ("match", "regex"),
) -> dict | list[object] | str | None:
) -> dict | list[Any] | str | None:
"""
Deep-copy `payload` and replace every configured string field with "[REDACTED]"
anywhere in nested dict/list structures.
@ -796,13 +796,13 @@ def set_response_cost_in_hidden_params(response: _CarriesHiddenParams, cost: flo
BATCH_PARENT_ID_KEY: Final = "batch_parent_id"
def is_batch_line_item_event(kwargs: object) -> bool:
if not isinstance(kwargs, Mapping):
return False
litellm_params: Final[object] = kwargs.get("litellm_params")
if not isinstance(litellm_params, Mapping):
return False
return bool(litellm_params.get(BATCH_PARENT_ID_KEY))
def is_batch_line_item_event(kwargs: Any) -> bool: # noqa: ANN401 # every hook passes its own unannotated callback kwargs
"""Whether a callback event is one JSONL line of a completed batch rather than a request"""
match kwargs:
case {"litellm_params": {"batch_parent_id": str() as batch_parent_id}}:
return bool(batch_parent_id)
case _:
return False
_HIDDEN_PARAMS_ADAPTER: Final = TypeAdapter(Mapping[str, object])

14
litellm/litellm_core_utils/litellm_logging.py
View file

@ -963,7 +963,7 @@ class Logging(LiteLLMLoggingBaseClass):
if model is not None:
self.model = model
self.user = user
marked_batch_parent_id: Final[object] = self.litellm_params.get(BATCH_PARENT_ID_KEY)
marked_batch_parent_id: Final[object] = self.litellm_params.get(BATCH_PARENT_ID_KEY) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # Logging.litellm_params is untyped upstream
self.litellm_params = {
**self.litellm_params,
**scrub_sensitive_keys_in_metadata(litellm_params),
@ -971,9 +971,9 @@ class Logging(LiteLLMLoggingBaseClass):
# Every spend, budget and metering hook skips an event carrying batch_parent_id, so only
# mark_batch_line_item may set it; the incoming params can hold caller request fields
if marked_batch_parent_id is None:
self.litellm_params.pop(BATCH_PARENT_ID_KEY, None)
self.litellm_params.pop(BATCH_PARENT_ID_KEY, None) # pyright: ignore[reportUnknownMemberType] # Logging.litellm_params is untyped upstream
else:
self.litellm_params[BATCH_PARENT_ID_KEY] = marked_batch_parent_id
self.litellm_params[BATCH_PARENT_ID_KEY] = marked_batch_parent_id # pyright: ignore[reportUnknownMemberType] # Logging.litellm_params is untyped upstream
self.litellm_request_debug = litellm_params.get("litellm_request_debug", False)
self.logger_fn = litellm_params.get("logger_fn", None)
if is_debugging_on() or self.litellm_request_debug:
@ -1013,7 +1013,7 @@ class Logging(LiteLLMLoggingBaseClass):
def mark_batch_line_item(self, batch_id: str) -> None:
"""Mark this logging object as one JSONL line of the completed batch ``batch_id``"""
self.litellm_params[BATCH_PARENT_ID_KEY] = batch_id
self.litellm_params[BATCH_PARENT_ID_KEY] = batch_id # pyright: ignore[reportUnknownMemberType] # Logging.litellm_params is untyped upstream
def update_from_kwargs(
self,
@ -3336,7 +3336,7 @@ class Logging(LiteLLMLoggingBaseClass):
cost_for_built_in_tools_cost_usd_dollar=0.0,
)
if batch_output_file_content is not None or batch_error_file_content is not None:
result_files = BatchResultFiles(
result_files = BatchResultFiles( # rebind-ok: the caller supplied the files
output=batch_output_file_content if isinstance(batch_output_file_content, bytes) else None,
error=batch_error_file_content if isinstance(batch_error_file_content, bytes) else None,
)
@ -3349,7 +3349,7 @@ class Logging(LiteLLMLoggingBaseClass):
litellm_params=self.litellm_params,
model_info=self.get_router_deployment_model_info(),
)
result_files = fetched_result_files
result_files = fetched_result_files # rebind-ok: the files this branch fetched
set_hidden_param(result, "response_cost", batch_result.cost)
set_hidden_param(result, "batch_models", batch_result.models)
@ -6582,7 +6582,7 @@ def _extract_response_obj_and_hidden_params(
original_exception, "_hidden_params", None
)
if isinstance(exception_hidden_params, dict) and exception_hidden_params:
hidden_params = dict(exception_hidden_params)
hidden_params = dict(exception_hidden_params) # rebind-ok: the exception carries the hidden params
elif (response_headers := _get_response_headers(original_exception)) is not None:
hidden_params = dict(
StandardLoggingHiddenParams(

2
litellm/llms/anthropic/chat/transformation.py
View file

@ -2887,7 +2887,7 @@ def _valid_user_id(user_id: str) -> bool:
def anthropic_message_to_model_response(result: Mapping[str, object], speed: str | None) -> ModelResponse:
pydantic_result: Final = AnthropicResponse.model_validate(result)
result_id: Final = result.get("id")
return AnthropicConfig().transform_parsed_response(
return AnthropicConfig().transform_parsed_response( # pyright: ignore[reportUnknownMemberType] # completion_response is a bare dict upstream
completion_response=pydantic_result.model_dump(),
raw_response=httpx.Response(
status_code=200,

2
litellm/llms/bedrock/batches/transformation.py
View file

@ -118,7 +118,7 @@ def bedrock_batch_line_to_response(
if "output" in model_output:
from ..chat.converse_transformation import AmazonConverseConfig
return AmazonConverseConfig()._transform_response( # pyright: ignore[reportPrivateUsage] # same reconstruction the converse chat path performs on the live response
return AmazonConverseConfig()._transform_response( # pyright: ignore[reportPrivateUsage, reportUnknownMemberType] # same reconstruction the converse chat path performs on the live response
model=model,
response=Response(200, json=dict(model_output)),
model_response=ModelResponse(),

7
litellm/proxy/proxy_server.py
View file

@ -7087,12 +7087,9 @@ class ProxyConfig:
health_check_concurrency = general_settings.get("health_check_concurrency", None)
health_check_details = general_settings.get("health_check_details", True)
### BATCH LINE ITEM CALLBACKS ###
_store_batch_line_items: Final[object] = general_settings.get("store_batch_line_items_in_callbacks")
_store_batch_line_items: Final = coerce_bool(general_settings.get("store_batch_line_items_in_callbacks")) # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # general_settings is untyped upstream
if _store_batch_line_items is not None:
if isinstance(_store_batch_line_items, str):
litellm.store_batch_line_items_in_callbacks = _store_batch_line_items.lower() == "true"
else:
litellm.store_batch_line_items_in_callbacks = bool(_store_batch_line_items)
litellm.store_batch_line_items_in_callbacks = bool(_store_batch_line_items)
# Health-check-driven routing (opt-in, passes through to Router later)
_enable_hc_routing = general_settings.get("enable_health_check_routing", False)
_hc_staleness = general_settings.get("health_check_staleness_threshold", None)

52
tests/unit/batches/test_batch_line_item_logging.py
View file

@ -144,9 +144,7 @@ def recorder():
litellm._async_success_callback
) # test-quality-ok: the feature dispatches through this global list; restored in teardown
saved_failure = list(litellm._async_failure_callback) # test-quality-ok: same dispatch seam, restored in teardown
litellm._async_success_callback = [
logger
] # test-quality-ok: there is no injection seam for callback lists; teardown restores
litellm._async_success_callback = [logger] # test-quality-ok: no injection seam; teardown restores
litellm._async_failure_callback = [logger] # test-quality-ok: same dispatch seam, restored in teardown
yield logger
litellm.store_batch_line_items_in_callbacks = saved_flag # test-quality-ok: teardown restoring the value set above
@ -195,9 +193,7 @@ def _hidden(event: dict) -> dict:
@pytest.mark.asyncio
async def test_line_items_emitted_alongside_aggregate(recorder):
litellm.store_batch_line_items_in_callbacks = (
True # test-quality-ok: the flag under test is a module global; fixture restores it
)
litellm.store_batch_line_items_in_callbacks = True # test-quality-ok: flag under test; fixture restores it
batch = _batch()
with (
patch(
@ -253,9 +249,7 @@ async def test_flag_off_emits_only_aggregate(recorder):
@pytest.mark.asyncio
async def test_line_items_skipped_for_unsupported_provider(recorder):
litellm.store_batch_line_items_in_callbacks = (
True # test-quality-ok: the flag under test is a module global; fixture restores it
)
litellm.store_batch_line_items_in_callbacks = True # test-quality-ok: flag under test; fixture restores it
file_mock: Final = AsyncMock(side_effect=_file_content)
with patch(
"litellm.files.main.afile_content", file_mock
@ -271,9 +265,7 @@ async def test_line_items_skipped_for_unsupported_provider(recorder):
@pytest.mark.asyncio
async def test_in_progress_batch_poll_emits_no_line_events(recorder):
litellm.store_batch_line_items_in_callbacks = (
True # test-quality-ok: the flag under test is a module global; fixture restores it
)
litellm.store_batch_line_items_in_callbacks = True # test-quality-ok: flag under test; fixture restores it
in_progress: Final = LiteLLMBatch(
id="batch_wip",
object="batch",
@ -298,9 +290,7 @@ async def test_in_progress_batch_poll_emits_no_line_events(recorder):
@pytest.mark.asyncio
async def test_input_fetch_failure_still_emits_aggregate(recorder):
litellm.store_batch_line_items_in_callbacks = (
True # test-quality-ok: the flag under test is a module global; fixture restores it
)
litellm.store_batch_line_items_in_callbacks = True # test-quality-ok: flag under test; fixture restores it
with patch(
"litellm.files.main.afile_content", new_callable=AsyncMock, side_effect=ValueError("boom")
): # test-quality-ok: afile_content is the provider boundary; no injection seam for managed file fetch
@ -547,9 +537,7 @@ def _parent_logging_with_params(litellm_params: dict) -> Logging:
@pytest.mark.asyncio
async def test_line_items_edge_shapes_and_edge_cases(recorder):
litellm.store_batch_line_items_in_callbacks = (
True # test-quality-ok: the flag under test is a module global; fixture restores it
)
litellm.store_batch_line_items_in_callbacks = True # test-quality-ok: flag under test; fixture restores it
batch = LiteLLMBatch(
id="batch_edge",
object="batch",
@ -611,9 +599,7 @@ async def test_line_items_edge_shapes_and_edge_cases(recorder):
@pytest.mark.asyncio
async def test_line_items_child_params_drop_parent_credentials(recorder):
litellm.store_batch_line_items_in_callbacks = (
True # test-quality-ok: the flag under test is a module global; fixture restores it
)
litellm.store_batch_line_items_in_callbacks = True # test-quality-ok: flag under test; fixture restores it
file_mock: Final = AsyncMock(side_effect=_file_content)
parent: Final = _parent_logging_with_params(
{
@ -648,9 +634,7 @@ async def test_line_items_child_params_drop_parent_credentials(recorder):
@pytest.mark.asyncio
async def test_line_items_anthropic_shapes(recorder):
litellm.store_batch_line_items_in_callbacks = (
True # test-quality-ok: the flag under test is a module global; fixture restores it
)
litellm.store_batch_line_items_in_callbacks = True # test-quality-ok: flag under test; fixture restores it
batch = LiteLLMBatch(
id="batch_anth",
object="batch",
@ -704,9 +688,7 @@ def _provider_batch(batch_id: str, input_file_id: str, output_file_id: str) -> L
@pytest.mark.asyncio
async def test_line_items_bedrock_shapes(recorder):
litellm.store_batch_line_items_in_callbacks = (
True # test-quality-ok: the flag under test is a module global; fixture restores it
)
litellm.store_batch_line_items_in_callbacks = True # test-quality-ok: flag under test; fixture restores it
file_mock: Final = AsyncMock(side_effect=_edge_file_content)
with (
patch(
@ -751,9 +733,7 @@ async def test_line_items_bedrock_shapes(recorder):
@pytest.mark.asyncio
async def test_line_items_mistral_shape(recorder):
litellm.store_batch_line_items_in_callbacks = (
True # test-quality-ok: the flag under test is a module global; fixture restores it
)
litellm.store_batch_line_items_in_callbacks = True # test-quality-ok: flag under test; fixture restores it
file_mock: Final = AsyncMock(side_effect=_edge_file_content)
with (
patch(
@ -820,9 +800,7 @@ def _scoped_file_content(file_map):
@pytest.mark.asyncio
async def test_line_items_anthropic_failure_keeps_provider_error(recorder):
litellm.store_batch_line_items_in_callbacks = (
True # test-quality-ok: the flag under test is a module global; fixture restores it
)
litellm.store_batch_line_items_in_callbacks = True # test-quality-ok: flag under test; fixture restores it
file_mock: Final = AsyncMock(
side_effect=_scoped_file_content(
{
@ -860,9 +838,7 @@ BODYLESS_ERR_OUTPUT_JSONL = json.dumps({"custom_id": "bare", "response": None, "
@pytest.mark.asyncio
async def test_line_items_failure_without_error_or_response_body_still_reaches_callbacks(recorder):
litellm.store_batch_line_items_in_callbacks = (
True # test-quality-ok: the flag under test is a module global; fixture restores it
)
litellm.store_batch_line_items_in_callbacks = True # test-quality-ok: flag under test; fixture restores it
file_mock: Final = AsyncMock(
side_effect=_scoped_file_content(
{
@ -889,9 +865,7 @@ async def test_line_items_failure_without_error_or_response_body_still_reaches_c
@pytest.mark.asyncio
async def test_line_items_native_vertex_rows_are_skipped(recorder):
litellm.store_batch_line_items_in_callbacks = (
True # test-quality-ok: the flag under test is a module global; fixture restores it
)
litellm.store_batch_line_items_in_callbacks = True # test-quality-ok: flag under test; fixture restores it
file_mock: Final = AsyncMock(
side_effect=_scoped_file_content(
{

8
tests/unit/integrations/test_batch_line_item_metering_sinks.py
View file

@ -149,8 +149,8 @@ async def _log_completed_batch(monkeypatch: pytest.MonkeyPatch, loggers: list, f
monkeypatch.setattr(litellm, "store_batch_line_items_in_callbacks", flag, raising=False)
saved_success = list(litellm._async_success_callback)
saved_failure = list(litellm._async_failure_callback)
litellm._async_success_callback = list(loggers)
litellm._async_failure_callback = list(loggers)
litellm._async_success_callback = list(loggers) # test-quality-ok: no injection seam; finally restores
litellm._async_failure_callback = list(loggers) # test-quality-ok: same dispatch seam
buf = io.StringIO()
try:
with (
@ -169,8 +169,8 @@ async def _log_completed_batch(monkeypatch: pytest.MonkeyPatch, loggers: list, f
batch_completion_cost=0.5,
)
finally:
litellm._async_success_callback = saved_success
litellm._async_failure_callback = saved_failure
litellm._async_success_callback = saved_success # test-quality-ok: restores the seam
litellm._async_failure_callback = saved_failure # test-quality-ok: restores the seam
def _openmeter_sinks() -> tuple[Any, RecordingHTTP]:

2
tests/unit/proxy/test_proxy_server_endpoints_and_startup.py
View file

@ -8153,7 +8153,7 @@ async def test_update_general_settings_store_batch_line_items_in_callbacks():
await proxy_config._update_general_settings(db_general_settings={})
assert litellm.store_batch_line_items_in_callbacks is False
finally:
litellm.store_batch_line_items_in_callbacks = saved_flag
litellm.store_batch_line_items_in_callbacks = saved_flag # test-quality-ok: restores the prior flag
@pytest.mark.asyncio