From b94613e7ae38aacfcf1063d6aa1621ea04fa5c11 Mon Sep 17 00:00:00 2001 From: Alex Castillo Date: Fri, 16 Oct 2026 14:35:57 -0400 Subject: [PATCH] fix(sap): validate the deployment URL override and configuration reads with pydantic The strict type gate counted the new optional_params and configuration dict reads as unknown types, and typing.cast is banned, so validate both at the boundary with small frozen models --- litellm/llms/sap/chat/transformation.py | 36 ++++++++++++++++++------- 1 file changed, 26 insertions(+), 10 deletions(-) diff --git a/litellm/llms/sap/chat/transformation.py b/litellm/llms/sap/chat/transformation.py index e05364c5993..3f275d0a2e7 100755 --- a/litellm/llms/sap/chat/transformation.py +++ b/litellm/llms/sap/chat/transformation.py @@ -9,7 +9,7 @@ from functools import cached_property from typing import TYPE_CHECKING, Any, Final, Union import httpx -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict, Field import litellm from litellm.types.llms.openai import AllMessageValues @@ -56,6 +56,21 @@ _SAP_MODEL_PARAMS_EXCLUDED_KEYS: Final[frozenset[str]] = frozenset( ) +class _DeploymentUrlOverride(BaseModel): + """The deployment_url a caller can pass in optional_params to skip discovery""" + + model_config = ConfigDict(frozen=True, extra="ignore") + deployment_url: str | None = None + + +class _OrchestrationConfiguration(BaseModel): + """The fields discovery reads from an AI Core /lm/configurations/{id} response""" + + model_config = ConfigDict(frozen=True, extra="ignore") + executable_id: str = Field(default="", alias="executableId") + name: str = "" + + def validate_dict( data: Mapping[str, object], model: type[BaseModel] ) -> dict: # mutable-ok: pydantic validation boundary; both input and output are untyped wire dicts @@ -190,12 +205,14 @@ class GenAIHubOrchestrationConfig(OpenAIGPTConfig): valid: Final[list[tuple[str, str, str]]] = [] for dep in deployments.get("resources", []): if dep.get("scenarioId") == "orchestration": - cfg = client.get( - f"{self.base_url}/lm/configurations/{dep['configurationId']}", - headers=self.headers, - ).json() - if cfg.get("executableId") == "orchestration": - valid.append((dep["deploymentUrl"], dep["createdAt"], cfg.get("name", ""))) + cfg = _OrchestrationConfiguration.model_validate( + client.get( + f"{self.base_url}/lm/configurations/{dep['configurationId']}", + headers=self.headers, + ).json() + ) + if cfg.executable_id == "orchestration": + valid.append((dep["deploymentUrl"], dep["createdAt"], cfg.name)) if not valid: raise GenAIHubOrchestrationError( status_code=400, @@ -284,10 +301,9 @@ class GenAIHubOrchestrationConfig(OpenAIGPTConfig): litellm_params: dict, stream: bool | None = None, ): + explicit_url: Final = _DeploymentUrlOverride.model_validate(optional_params).deployment_url deployment_url: Final = ( - optional_params.get("deployment_url") - or os.environ.get("AICORE_ORCHESTRATION_DEPLOYMENT_URL") - or self.deployment_url + explicit_url or os.environ.get("AICORE_ORCHESTRATION_DEPLOYMENT_URL") or self.deployment_url ) return f"{deployment_url.rstrip('/')}/v2/completion"