fix(openrouter): comply with type discipline gate requirements

This commit is contained in:
yyouretoast 2026-10-16 16:46:14 +03:00
parent b7530315f8
commit 59d26ba6b6
5 changed files with 65 additions and 50 deletions

View file

@ -2225,15 +2225,17 @@ def _map_openrouter_exception(
)
elif original_exception.status_code == 404:
if "supporting your data region" in error_str.lower():
response_obj = getattr(original_exception, "response", None)
inferred_region = None
if response_obj is not None and getattr(response_obj, "url", None):
from litellm.llms.openrouter.data_residency import infer_openrouter_data_residency
response_obj: Final = getattr(original_exception, "response", None)
from litellm.llms.openrouter.data_residency import infer_openrouter_data_residency
inferred_region = infer_openrouter_data_residency(
inferred_region: Final[str | None] = (
infer_openrouter_data_residency(
custom_llm_provider="openrouter",
api_base=str(response_obj.url),
)
if response_obj is not None and getattr(response_obj, "url", None)
else None
)
raise OpenRouterRegionNotSupportedError(
message=f"OpenRouterRegionNotSupportedError: {exception_provider} - {error_str}",
model=model,

View file

@ -52,31 +52,29 @@ class OpenrouterConfig(OpenAIGPTConfig):
def __init__(
self,
region: str | None = None,
**kwargs: object,
) -> None:
self.region = region
super().__init__(**kwargs)
super().__init__()
def get_complete_url(
self,
api_base: str | None,
api_key: str | None,
model: str,
optional_params: dict,
litellm_params: dict,
optional_params: Mapping[str, object] | None = None,
litellm_params: Mapping[str, object] | None = None,
stream: bool | None = None,
) -> str:
api_base = resolve_openrouter_api_base(
resolved_base: Final = resolve_openrouter_api_base(
api_base=api_base,
region=self.region,
litellm_params=litellm_params,
optional_params=optional_params,
)
).rstrip("/")
endpoint: Final = "chat/completions"
api_base = api_base.rstrip("/")
if endpoint in api_base:
return api_base
return f"{api_base}/{endpoint}"
if endpoint in resolved_base:
return resolved_base
return f"{resolved_base}/{endpoint}"
def get_supported_openai_params(self, model: str) -> list:
"""

View file

@ -9,6 +9,8 @@ OpenRouter provides dedicated regional endpoints for end-to-end data residency:
Docs: https://openrouter.ai/docs/guides/features/in-region-routing
"""
from collections.abc import Mapping, MutableMapping
from types import MappingProxyType
from typing import Final
from urllib.parse import urlparse
@ -16,19 +18,23 @@ import litellm
from litellm.secret_managers.main import get_secret_str
# Normalized region names mapped to their dedicated OpenRouter base URLs
_OPENROUTER_REGIONAL_URLS: Final[dict[str, str]] = {
"eu": "https://eu.openrouter.ai/api/v1",
"europe": "https://eu.openrouter.ai/api/v1",
"us": "https://us.openrouter.ai/api/v1",
"usa": "https://us.openrouter.ai/api/v1",
"united states": "https://us.openrouter.ai/api/v1",
}
_OPENROUTER_REGIONAL_URLS: Final[Mapping[str, str]] = MappingProxyType(
{
"eu": "https://eu.openrouter.ai/api/v1",
"europe": "https://eu.openrouter.ai/api/v1",
"us": "https://us.openrouter.ai/api/v1",
"usa": "https://us.openrouter.ai/api/v1",
"united states": "https://us.openrouter.ai/api/v1",
}
)
# OpenRouter regional hostnames mapped to canonical data residency codes ("eu", "us")
_OPENROUTER_REGIONAL_HOSTNAMES: Final[dict[str, str]] = {
"eu.openrouter.ai": "eu",
"us.openrouter.ai": "us",
}
_OPENROUTER_REGIONAL_HOSTNAMES: Final[Mapping[str, str]] = MappingProxyType(
{
"eu.openrouter.ai": "eu",
"us.openrouter.ai": "us",
}
)
_DEFAULT_OPENROUTER_API_BASE: Final[str] = "https://openrouter.ai/api/v1"
@ -36,8 +42,8 @@ _DEFAULT_OPENROUTER_API_BASE: Final[str] = "https://openrouter.ai/api/v1"
def resolve_openrouter_api_base(
api_base: str | None = None,
region: str | None = None,
litellm_params: dict | None = None,
optional_params: dict | None = None,
litellm_params: Mapping[str, object] | None = None,
optional_params: Mapping[str, object] | None = None,
) -> str:
"""
Resolve the OpenRouter API base URL based on custom api_base, explicit region, or environment.
@ -48,34 +54,44 @@ def resolve_openrouter_api_base(
3. Fallback to `api_base or litellm.api_base or OPENROUTER_API_BASE or default global URL`.
"""
# Check if a custom api_base was provided that is not the default global endpoint
is_custom_api_base = False
if api_base:
cleaned_base = api_base.rstrip("/").lower()
if cleaned_base not in ("https://openrouter.ai/api/v1", "https://openrouter.ai", "http://openrouter.ai"):
is_custom_api_base = True
cleaned_base: Final[str | None] = api_base.rstrip("/").lower() if api_base else None
is_custom_api_base: Final[bool] = bool(
cleaned_base
and cleaned_base not in ("https://openrouter.ai/api/v1", "https://openrouter.ai", "http://openrouter.ai")
)
if is_custom_api_base and api_base:
return api_base
# Check for region in litellm_params, optional_params, explicit region argument, config, or environment
resolved_region = None
if isinstance(litellm_params, dict):
resolved_region = (
litellm_params.get("region") or litellm_params.get("region_name") or litellm_params.get("data_residency")
)
if not resolved_region and isinstance(optional_params, dict):
resolved_region = (
raw_from_params: Final[object] = (
litellm_params.get("region") or litellm_params.get("region_name") or litellm_params.get("data_residency")
if isinstance(litellm_params, Mapping)
else None
)
raw_from_optional: Final[object] = (
(
optional_params.pop("region", None)
or optional_params.pop("region_name", None)
or optional_params.pop("data_residency", None)
)
if not resolved_region:
resolved_region = (
region or getattr(litellm, "openrouter_region", None) or getattr(litellm, "openrouter_data_residency", None)
if isinstance(optional_params, MutableMapping)
else (
optional_params.get("region") or optional_params.get("region_name") or optional_params.get("data_residency")
if isinstance(optional_params, Mapping)
else None
)
)
resolved_region: Final[object] = (
raw_from_params
or raw_from_optional
or region
or getattr(litellm, "openrouter_region", None)
or getattr(litellm, "openrouter_data_residency", None)
)
if resolved_region and isinstance(resolved_region, str):
normalized_region = resolved_region.strip().lower()
normalized_region: Final[str] = resolved_region.strip().lower()
if normalized_region in ("global", "default"):
return _DEFAULT_OPENROUTER_API_BASE
if normalized_region in _OPENROUTER_REGIONAL_URLS:

View file

@ -85,15 +85,14 @@ class OpenrouterEmbeddingConfig(BaseEmbeddingConfig):
"""
Get the complete URL for OpenRouter Embedding API endpoint.
"""
api_base = resolve_openrouter_api_base(
resolved_base: Final = resolve_openrouter_api_base(
api_base=api_base,
litellm_params=litellm_params,
optional_params=optional_params,
)
api_base = api_base.rstrip("/")
).rstrip("/")
# Return the embeddings endpoint
return f"{api_base}/embeddings"
return f"{resolved_base}/embeddings"
def transform_embedding_request(
self,

View file

@ -3415,7 +3415,7 @@ def _complete_openrouter(ctx: CompletionDispatchContext) -> _CompletionDispatchR
## Load Config
config: Final = litellm.OpenrouterConfig.get_config()
api_base = resolve_openrouter_api_base(
api_base = resolve_openrouter_api_base( # rebind-ok: regional endpoint resolution for openrouter
api_base=api_base,
region=getattr(config, "region", None),
litellm_params=litellm_params,
@ -6788,7 +6788,7 @@ def embedding(
headers=headers,
)
elif custom_llm_provider == "openrouter":
api_base = resolve_openrouter_api_base(
api_base = resolve_openrouter_api_base( # rebind-ok: regional endpoint resolution for openrouter
api_base=api_base,
litellm_params=litellm_params_dict,
optional_params=optional_params,