From e00471b86a74acb7ab97efe3095617a727bca651 Mon Sep 17 00:00:00 2001 From: yucheng-berri Date: Thu, 8 Oct 2026 15:08:46 -0700 Subject: [PATCH] chore(batches): keep the lint gates within main's ceilings after the merge 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. --- .../proxy/common_utils/check_batch_cost.py | 2 +- litellm/batches/batch_line_item_logging.py | 14 ++--- litellm/batches/batch_utils.py | 4 +- litellm/litellm_core_utils/core_helpers.py | 24 ++++----- litellm/litellm_core_utils/litellm_logging.py | 14 ++--- litellm/llms/anthropic/chat/transformation.py | 2 +- .../llms/bedrock/batches/transformation.py | 2 +- litellm/proxy/proxy_server.py | 7 +-- .../batches/test_batch_line_item_logging.py | 52 +++++-------------- .../test_batch_line_item_metering_sinks.py | 8 +-- ...test_proxy_server_endpoints_and_startup.py | 2 +- 11 files changed, 51 insertions(+), 80 deletions(-) diff --git a/enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py b/enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py index 167f4dac82c..c885fde6ae4 100644 --- a/enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py +++ b/enterprise/litellm_enterprise/proxy/common_utils/check_batch_cost.py @@ -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: diff --git a/litellm/batches/batch_line_item_logging.py b/litellm/batches/batch_line_item_logging.py index 9565614f78f..854f8c2ef34 100644 --- a/litellm/batches/batch_line_item_logging.py +++ b/litellm/batches/batch_line_item_logging.py @@ -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) diff --git a/litellm/batches/batch_utils.py b/litellm/batches/batch_utils.py index 26ae92526e0..38736b43224 100644 --- a/litellm/batches/batch_utils.py +++ b/litellm/batches/batch_utils.py @@ -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: diff --git a/litellm/litellm_core_utils/core_helpers.py b/litellm/litellm_core_utils/core_helpers.py index 0a18fc19cd1..8ffcc4939dc 100644 --- a/litellm/litellm_core_utils/core_helpers.py +++ b/litellm/litellm_core_utils/core_helpers.py @@ -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]) diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 923059d01b9..26d9035e7a2 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -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( diff --git a/litellm/llms/anthropic/chat/transformation.py b/litellm/llms/anthropic/chat/transformation.py index 0633280ebf4..f65cfcb18fa 100644 --- a/litellm/llms/anthropic/chat/transformation.py +++ b/litellm/llms/anthropic/chat/transformation.py @@ -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, diff --git a/litellm/llms/bedrock/batches/transformation.py b/litellm/llms/bedrock/batches/transformation.py index 51608ad3a05..136cf8f1b99 100644 --- a/litellm/llms/bedrock/batches/transformation.py +++ b/litellm/llms/bedrock/batches/transformation.py @@ -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(), diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 707ab48219b..f07656bc51d 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -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) diff --git a/tests/unit/batches/test_batch_line_item_logging.py b/tests/unit/batches/test_batch_line_item_logging.py index 43da11528b0..d383546b7bf 100644 --- a/tests/unit/batches/test_batch_line_item_logging.py +++ b/tests/unit/batches/test_batch_line_item_logging.py @@ -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( { diff --git a/tests/unit/integrations/test_batch_line_item_metering_sinks.py b/tests/unit/integrations/test_batch_line_item_metering_sinks.py index 525f6ae15d0..9cec64b1273 100644 --- a/tests/unit/integrations/test_batch_line_item_metering_sinks.py +++ b/tests/unit/integrations/test_batch_line_item_metering_sinks.py @@ -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]: diff --git a/tests/unit/proxy/test_proxy_server_endpoints_and_startup.py b/tests/unit/proxy/test_proxy_server_endpoints_and_startup.py index 038c991040f..0b0708955bb 100644 --- a/tests/unit/proxy/test_proxy_server_endpoints_and_startup.py +++ b/tests/unit/proxy/test_proxy_server_endpoints_and_startup.py @@ -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