linting issues fixed

This commit is contained in:
Yamac Ay 2026-10-29 13:31:32 +01:00
parent 88a212f500
commit 87cb2f5b28
No known key found for this signature in database
GPG key ID: D113B438819CA628
3 changed files with 18 additions and 18 deletions

View file

@ -42,7 +42,7 @@ from .models import (
SAPUserMessage,
)
_AICORE_ORCHESTRATION_DEPLOYMENT_URL_ENV_VAR = "AICORE_ORCHESTRATION_DEPLOYMENT_URL"
_AICORE_ORCHESTRATION_DEPLOYMENT_URL_ENV_VAR: Final = "AICORE_ORCHESTRATION_DEPLOYMENT_URL"
# Keys routed outside SAP orchestration `model.params` (prompt, stream, fallbacks, etc.)
_SAP_MODEL_PARAMS_EXCLUDED_KEYS: Final[frozenset[str]] = frozenset(
@ -263,7 +263,7 @@ class GenAIHubOrchestrationConfig(OpenAIGPTConfig):
stream: bool | None = None,
):
# Per-request override wins; deployment_url handles env var + discovery.
base = optional_params.get("deployment_url") or self.deployment_url
base: Final = optional_params.get("deployment_url") or self.deployment_url
return join_api_base(base, "v2/completion")
def _build_prompt_module(

View file

@ -17,7 +17,9 @@ from .chat.handler import GenAIHubOrchestrationError
from .credentials import AICoreDeploymentList
from litellm.litellm_core_utils.url_utils import join_api_base
_deployment_url_cache: dict[tuple[str, str], str] = {} # mutable-ok: process-wide cache, guarded by _deployment_url_lock
_deployment_url_cache: dict[ # mutable-ok: process-wide cache, guarded by _deployment_url_lock # rebind-ok: once initialized
tuple[str, str], str
] = {}
_deployment_url_lock: Final = threading.Lock()
@ -25,19 +27,17 @@ def fetch_deployment_url(
base_url: str,
headers: Mapping[str, str],
) -> str:
deployments: Final[AICoreDeploymentList] = (
litellm.module_level_client.get(
join_api_base(base_url, "lm/deployments"),
headers=headers,
params={ # mutable-ok: one-shot query-params dict passed directly to httpx, not stored
"scenarioId": "orchestration",
"executableIds": ["orchestration"],
"status": "RUNNING",
},
).json()
)
deployments: Final[AICoreDeploymentList] = litellm.module_level_client.get(
join_api_base(base_url, "lm/deployments"),
headers=headers,
params={
"scenarioId": "orchestration",
"executableIds": ["orchestration"],
"status": "RUNNING",
},
).json()
candidates: Final[list[tuple[str, str, str]]] = sorted(
candidates: Final[list[tuple[str, str, str]]] = sorted( # mutable-ok: list of deployment candidates
(
(resource["deploymentUrl"], resource["createdAt"], resource["configurationName"])
for resource in deployments.get("resources", [])
@ -66,11 +66,11 @@ def resolve_deployment_url(
fetcher: Callable[[], str],
) -> str:
cache_key: Final[tuple[str, str]] = (base_url, resource_group)
cached_url = _deployment_url_cache.get(cache_key)
cached_url = _deployment_url_cache.get(cache_key) # rebind-ok: fast-path read before acquiring the lock
if cached_url is not None:
return cached_url
with _deployment_url_lock:
rechecked_url = _deployment_url_cache.get(cache_key)
rechecked_url = _deployment_url_cache.get(cache_key) # rebind-ok: double-check under lock after fast-path miss
if rechecked_url is not None:
return rechecked_url
fetched_url: Final[str] = os.environ.get(env_var) or fetcher()

View file

@ -17,7 +17,7 @@ from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.llms.openai import AllEmbeddingInputValues
from litellm.types.utils import EmbeddingResponse
_AICORE_EMBEDDING_DEPLOYMENT_URL_ENV_VAR = "AICORE_EMBEDDING_DEPLOYMENT_URL"
_AICORE_EMBEDDING_DEPLOYMENT_URL_ENV_VAR: Final = "AICORE_EMBEDDING_DEPLOYMENT_URL"
from ..chat.handler import GenAIHubOrchestrationError
from ..credentials import get_token_creator