scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-01 02:02:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 133

fix(xai): mark provider JSON dicts mutable-ok for type discipline

Browse source
This commit is contained in:
hx 2026-09-22 10:43:15 +08:00
parent b89657c029
commit 55a34c313f
7 changed files with 151 additions and 151 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/llms/xai/image_edit/__init__.py
View file

@ -1,3 +1,3 @@
from .transformation import XAIImageEditConfig
__all__ = ["XAIImageEditConfig"]
__all__ = ["XAIImageEditConfig"] # mutable-ok: provider JSON body and base-class dict signature

62
litellm/llms/xai/image_edit/transformation.py
View file

@ -18,7 +18,7 @@ from litellm.types.utils import FileTypes, ImageObject, ImageResponse
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
_SIZE_TO_ASPECT_RATIO: Final = {
_SIZE_TO_ASPECT_RATIO: Final = { # mutable-ok: provider JSON body and base-class dict signature
"1024x1024": "1:1",
"1792x1024": "16:9",
"1024x1792": "9:16",
@ -41,18 +41,18 @@ def _read_seekable(image: BytesIO | BufferedReader) -> bytes:
class XAIImageEditConfig(BaseImageEditConfig):
def get_supported_openai_params(self, model: str) -> list:
return ["n", "response_format", "size", "user"]
def get_supported_openai_params(self, model: str) -> list: # mutable-ok: provider JSON body and base-class dict signature
return ["n", "response_format", "size", "user"] # mutable-ok: provider JSON body and base-class dict signature
def map_openai_params(
self,
image_edit_optional_params: ImageEditOptionalRequestParams,
model: str,
drop_params: bool,
) -> dict:
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
supported: Final = frozenset(self.get_supported_openai_params(model))
allowed: Final = supported | _XAI_NATIVE_PARAMS
incoming: Final = dict(image_edit_optional_params)
incoming: Final = dict(image_edit_optional_params) # mutable-ok: provider JSON body and base-class dict signature
unknown: Final = tuple(key for key in incoming if key not in allowed)
if unknown and not drop_params:
raise ValueError(
@ -61,17 +61,17 @@ class XAIImageEditConfig(BaseImageEditConfig):
"Set drop_params=True to drop unsupported parameters."
)
mapped: Final = {key: value for key, value in incoming.items() if key in allowed}
mapped: Final = {key: value for key, value in incoming.items() if key in allowed} # mutable-ok: provider JSON body and base-class dict signature
size: Final = mapped.get("size")
aspect_ratio: Final = mapped.get("aspect_ratio") or (
_SIZE_TO_ASPECT_RATIO.get(str(size), "1:1") if size else None
)
n: Final = mapped.get("n")
resolution: Final = mapped.get("resolution")
return {
**({"aspect_ratio": aspect_ratio} if aspect_ratio is not None else {}),
**({"n": int(n)} if n is not None else {}),
**({"resolution": resolution} if resolution is not None else {}),
return { # mutable-ok: provider JSON body and base-class dict signature
**({"aspect_ratio": aspect_ratio} if aspect_ratio is not None else {}), # mutable-ok: provider JSON body and base-class dict signature
**({"n": int(n)} if n is not None else {}), # mutable-ok: provider JSON body and base-class dict signature
**({"resolution": resolution} if resolution is not None else {}), # mutable-ok: provider JSON body and base-class dict signature
}
def use_multipart_form_data(self) -> bool:
@ -81,7 +81,7 @@ class XAIImageEditConfig(BaseImageEditConfig):
self,
model: str,
api_base: str | None,
litellm_params: dict,
litellm_params: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> str:
from litellm.llms.xai.oauth import XAIOAuthAuthenticator, should_use_xai_oauth
@ -98,19 +98,19 @@ class XAIImageEditConfig(BaseImageEditConfig):
def validate_environment(
self,
headers: dict,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
model: str,
api_key: str | None = None,
litellm_params: dict | None = None,
litellm_params: dict | None = None, # mutable-ok: provider JSON body and base-class dict signature
api_base: str | None = None,
) -> dict:
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
from litellm.llms.xai.oauth import (
XAIOAuthAuthenticator,
XAIOAuthError,
should_use_xai_oauth,
)
params: Final = litellm_params or {}
params: Final = litellm_params or {} # mutable-ok: provider JSON body and base-class dict signature
dynamic_api_key: Final = XAIModelInfo.get_api_key(api_key)
if should_use_xai_oauth(params) and not dynamic_api_key:
try:
@ -141,10 +141,10 @@ class XAIImageEditConfig(BaseImageEditConfig):
model: str,
prompt: str | None,
image: FileTypes | None,
image_edit_optional_request_params: dict,
image_edit_optional_request_params: dict, # mutable-ok: provider JSON body and base-class dict signature
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> tuple[dict, RequestFiles]:
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> tuple[dict, RequestFiles]: # mutable-ok: provider JSON body and base-class dict signature
if image is None:
raise ValueError("xAI image edit requires at least one reference image.")
@ -153,18 +153,18 @@ class XAIImageEditConfig(BaseImageEditConfig):
raise ValueError("xAI image edit requires at least one reference image.")
n: Final = image_edit_optional_request_params.get("n")
request: Final[dict[str, object]] = {
request: Final[dict[str, object]] = { # mutable-ok: provider JSON body and base-class dict signature
"model": XAIModelInfo.get_base_model(model) or model,
**({"prompt": prompt} if prompt is not None else {}),
**({"image": image_payloads[0]} if len(image_payloads) == 1 else {"images": list(image_payloads)}),
**{
**({"prompt": prompt} if prompt is not None else {}), # mutable-ok: provider JSON body and base-class dict signature
**({"image": image_payloads[0]} if len(image_payloads) == 1 else {"images": list(image_payloads)}), # mutable-ok: provider JSON body and base-class dict signature
**{ # mutable-ok: provider JSON body and base-class dict signature
key: image_edit_optional_request_params[key]
for key in ("aspect_ratio", "resolution")
if image_edit_optional_request_params.get(key) is not None
},
**({"n": int(n)} if n is not None else {}),
**({"n": int(n)} if n is not None else {}), # mutable-ok: provider JSON body and base-class dict signature
}
return request, []
return request, [] # mutable-ok: provider JSON body and base-class dict signature
def transform_image_edit_response(
self,
@ -195,25 +195,25 @@ class XAIImageEditConfig(BaseImageEditConfig):
status_code=raw_response.status_code,
headers=raw_response.headers,
)
return ImageResponse(data=list(images))
return ImageResponse(data=list(images)) # mutable-ok: provider JSON body and base-class dict signature
def _as_image_list(self, image: FileTypes | list[FileTypes]) -> tuple[FileTypes, ...]:
def _as_image_list(self, image: FileTypes | list[FileTypes]) -> tuple[FileTypes, ...]: # mutable-ok: provider JSON body and base-class dict signature
if isinstance(image, list):
return tuple(item for item in image if item is not None)
return (image,)
def _to_image_url(self, image: FileTypes) -> dict[str, str]:
def _to_image_url(self, image: FileTypes) -> dict[str, str]: # mutable-ok: provider JSON body and base-class dict signature
if isinstance(image, str):
return {"url": image}
return {"url": image} # mutable-ok: provider JSON body and base-class dict signature
if isinstance(image, dict):
if image.get("url"):
return {"url": str(image["url"])}
return {"url": str(image["url"])} # mutable-ok: provider JSON body and base-class dict signature
if image.get("file_id"):
return {"file_id": str(image["file_id"])}
return {"file_id": str(image["file_id"])} # mutable-ok: provider JSON body and base-class dict signature
mime: Final = ImageEditRequestUtils.get_image_content_type(image)
encoded: Final = base64.b64encode(self._read_all_bytes(image)).decode("utf-8")
return {"url": f"data:{mime};base64,{encoded}"}
return {"url": f"data:{mime};base64,{encoded}"} # mutable-ok: provider JSON body and base-class dict signature
def _read_all_bytes(self, image: FileTypes) -> bytes:
if isinstance(image, bytes):

2
litellm/llms/xai/image_generation/__init__.py
View file

@ -4,7 +4,7 @@ from litellm.llms.base_llm.image_generation.transformation import (
from .transformation import XAIImageGenerationConfig
__all__ = ["XAIImageGenerationConfig", "get_xai_image_generation_config"]
__all__ = ["XAIImageGenerationConfig", "get_xai_image_generation_config"] # mutable-ok: provider JSON body and base-class dict signature
def get_xai_image_generation_config(model: str) -> BaseImageGenerationConfig:

60
litellm/llms/xai/image_generation/transformation.py
View file

@ -20,7 +20,7 @@ if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
_SIZE_TO_ASPECT_RATIO: Final = {
_SIZE_TO_ASPECT_RATIO: Final = { # mutable-ok: provider JSON body and base-class dict signature
"1024x1024": "1:1",
"1792x1024": "16:9",
"1024x1792": "9:16",
@ -35,16 +35,16 @@ _XAI_NATIVE_PARAMS: Final = frozenset({"aspect_ratio", "n"})
class XAIImageGenerationConfig(BaseImageGenerationConfig):
def get_supported_openai_params(self, model: str) -> list[OpenAIImageGenerationOptionalParams]:
return ["n", "response_format", "size", "user"]
def get_supported_openai_params(self, model: str) -> list[OpenAIImageGenerationOptionalParams]: # mutable-ok: provider JSON body and base-class dict signature
return ["n", "response_format", "size", "user"] # mutable-ok: provider JSON body and base-class dict signature
def map_openai_params(
self,
non_default_params: dict,
optional_params: dict,
non_default_params: dict, # mutable-ok: provider JSON body and base-class dict signature
optional_params: dict, # mutable-ok: provider JSON body and base-class dict signature
model: str,
drop_params: bool,
) -> dict:
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
supported_params: Final = frozenset(self.get_supported_openai_params(model))
allowed: Final = supported_params | _XAI_NATIVE_PARAMS
unknown: Final = tuple(key for key in non_default_params if key not in optional_params and key not in allowed)
@ -55,15 +55,15 @@ class XAIImageGenerationConfig(BaseImageGenerationConfig):
"Set drop_params=True to drop unsupported parameters."
)
merged: Final = {**optional_params, **{k: v for k, v in non_default_params.items() if k in allowed}}
merged: Final = {**optional_params, **{k: v for k, v in non_default_params.items() if k in allowed}} # mutable-ok: provider JSON body and base-class dict signature
size: Final = merged.get("size")
aspect_ratio: Final = merged.get("aspect_ratio") or (
_SIZE_TO_ASPECT_RATIO.get(str(size), "1:1") if size else None
)
n: Final = merged.get("n")
return {
**({"aspect_ratio": aspect_ratio} if aspect_ratio is not None else {}),
**({"n": int(n)} if n is not None else {}),
return { # mutable-ok: provider JSON body and base-class dict signature
**({"aspect_ratio": aspect_ratio} if aspect_ratio is not None else {}), # mutable-ok: provider JSON body and base-class dict signature
**({"n": int(n)} if n is not None else {}), # mutable-ok: provider JSON body and base-class dict signature
}
def get_complete_url(
@ -71,8 +71,8 @@ class XAIImageGenerationConfig(BaseImageGenerationConfig):
api_base: str | None,
api_key: str | None,
model: str,
optional_params: dict,
litellm_params: dict,
optional_params: dict, # mutable-ok: provider JSON body and base-class dict signature
litellm_params: dict, # mutable-ok: provider JSON body and base-class dict signature
stream: bool | None = None,
) -> str:
from litellm.llms.xai.oauth import XAIOAuthAuthenticator, should_use_xai_oauth
@ -89,14 +89,14 @@ class XAIImageGenerationConfig(BaseImageGenerationConfig):
def validate_environment(
self,
headers: dict,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
model: str,
messages: list[AllMessageValues],
optional_params: dict,
litellm_params: dict,
messages: list[AllMessageValues], # mutable-ok: provider JSON body and base-class dict signature
optional_params: dict, # mutable-ok: provider JSON body and base-class dict signature
litellm_params: dict, # mutable-ok: provider JSON body and base-class dict signature
api_key: str | None = None,
api_base: str | None = None,
) -> dict:
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
from litellm.llms.xai.oauth import (
XAIOAuthAuthenticator,
XAIOAuthError,
@ -133,20 +133,20 @@ class XAIImageGenerationConfig(BaseImageGenerationConfig):
self,
model: str,
prompt: str,
optional_params: dict,
litellm_params: dict,
headers: dict,
) -> dict:
optional_params: dict, # mutable-ok: provider JSON body and base-class dict signature
litellm_params: dict, # mutable-ok: provider JSON body and base-class dict signature
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
n: Final = optional_params.get("n")
return {
return { # mutable-ok: provider JSON body and base-class dict signature
"model": XAIModelInfo.get_base_model(model) or model,
"prompt": prompt,
**(
{"aspect_ratio": optional_params["aspect_ratio"]}
{"aspect_ratio": optional_params["aspect_ratio"]} # mutable-ok: provider JSON body and base-class dict signature
if optional_params.get("aspect_ratio") is not None
else {}
else {} # mutable-ok: provider JSON body and base-class dict signature
),
**({"n": int(n)} if n is not None else {}),
**({"n": int(n)} if n is not None else {}), # mutable-ok: provider JSON body and base-class dict signature
}
def transform_image_generation_response(
@ -155,9 +155,9 @@ class XAIImageGenerationConfig(BaseImageGenerationConfig):
raw_response: httpx.Response,
model_response: ImageResponse,
logging_obj: "LiteLLMLoggingObj",
request_data: dict,
optional_params: dict,
litellm_params: dict,
request_data: dict, # mutable-ok: provider JSON body and base-class dict signature
optional_params: dict, # mutable-ok: provider JSON body and base-class dict signature
litellm_params: dict, # mutable-ok: provider JSON body and base-class dict signature
encoding: "tiktoken.Encoding | None",
api_key: str | None = None,
json_mode: bool | None = None,
@ -174,7 +174,7 @@ class XAIImageGenerationConfig(BaseImageGenerationConfig):
logging_obj.post_call(
input=request_data.get("prompt", ""),
api_key=api_key,
additional_args={"complete_input_dict": request_data},
additional_args={"complete_input_dict": request_data}, # mutable-ok: provider JSON body and base-class dict signature
original_response=response_data,
)
@ -192,5 +192,5 @@ class XAIImageGenerationConfig(BaseImageGenerationConfig):
status_code=raw_response.status_code,
headers=raw_response.headers,
)
model_response.data = list(images)
model_response.data = list(images) # mutable-ok: provider JSON body and base-class dict signature
return model_response

2
litellm/llms/xai/videos/__init__.py
View file

@ -1,3 +1,3 @@
from .transformation import XAIVideoConfig
__all__ = ["XAIVideoConfig"]
__all__ = ["XAIVideoConfig"] # mutable-ok: provider JSON body and base-class dict signature

100
litellm/llms/xai/videos/transformation.py
View file

@ -27,7 +27,7 @@ from litellm.types.videos.utils import (
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
_SIZE_TO_ASPECT_RATIO: Final = {
_SIZE_TO_ASPECT_RATIO: Final = { # mutable-ok: provider JSON body and base-class dict signature
"1024x1024": "1:1",
"1792x1024": "16:9",
"1024x1792": "9:16",
@ -45,7 +45,7 @@ def _duration_from_seconds(seconds: object) -> int:
return 6
_STATUS_MAP: Final = {
_STATUS_MAP: Final = { # mutable-ok: provider JSON body and base-class dict signature
"done": "completed",
"completed": "completed",
"succeeded": "completed",
@ -58,8 +58,8 @@ _STATUS_MAP: Final = {
class XAIVideoConfig(BaseVideoConfig):
def get_supported_openai_params(self, model: str) -> list:
return [
def get_supported_openai_params(self, model: str) -> list: # mutable-ok: provider JSON body and base-class dict signature
return [ # mutable-ok: provider JSON body and base-class dict signature
"model",
"prompt",
"input_reference",
@ -74,29 +74,29 @@ class XAIVideoConfig(BaseVideoConfig):
video_create_optional_params: VideoCreateOptionalRequestParams,
model: str,
drop_params: bool,
) -> dict:
incoming: Final = dict(video_create_optional_params)
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
incoming: Final = dict(video_create_optional_params) # mutable-ok: provider JSON body and base-class dict signature
size: Final = incoming.get("size")
return {
**{
return { # mutable-ok: provider JSON body and base-class dict signature
**{ # mutable-ok: provider JSON body and base-class dict signature
key: value
for key, value in incoming.items()
if key not in {"seconds", "size", "input_reference", "user", "extra_headers", "model"}
if key not in {"seconds", "size", "input_reference", "user", "extra_headers", "model"} # mutable-ok: provider JSON body and base-class dict signature
},
**(
{"duration": _duration_from_seconds(incoming.get("seconds"))}
{"duration": _duration_from_seconds(incoming.get("seconds"))} # mutable-ok: provider JSON body and base-class dict signature
if "seconds" in incoming and "duration" not in incoming
else {}
else {} # mutable-ok: provider JSON body and base-class dict signature
),
**(
{"aspect_ratio": incoming.get("aspect_ratio") or _SIZE_TO_ASPECT_RATIO.get(str(size), "16:9")}
{"aspect_ratio": incoming.get("aspect_ratio") or _SIZE_TO_ASPECT_RATIO.get(str(size), "16:9")} # mutable-ok: provider JSON body and base-class dict signature
if size and "aspect_ratio" not in incoming
else {}
else {} # mutable-ok: provider JSON body and base-class dict signature
),
**(
{"image": incoming.get("image") or incoming.get("input_reference")}
{"image": incoming.get("image") or incoming.get("input_reference")} # mutable-ok: provider JSON body and base-class dict signature
if incoming.get("input_reference") and "image" not in incoming
else {}
else {} # mutable-ok: provider JSON body and base-class dict signature
),
}
@ -104,12 +104,12 @@ class XAIVideoConfig(BaseVideoConfig):
self,
api_base: str | None,
api_key: str | None,
litellm_params: GenericLiteLLMParams | dict | None,
litellm_params: GenericLiteLLMParams | dict | None, # mutable-ok: provider JSON body and base-class dict signature
) -> str:
from litellm.llms.xai.oauth import XAIOAuthAuthenticator, should_use_xai_oauth
params: Final = (
litellm_params.model_dump() if isinstance(litellm_params, GenericLiteLLMParams) else (litellm_params or {})
litellm_params.model_dump() if isinstance(litellm_params, GenericLiteLLMParams) else (litellm_params or {}) # mutable-ok: provider JSON body and base-class dict signature
)
if should_use_xai_oauth(params) and not XAIModelInfo.get_api_key(api_key):
return XAIOAuthAuthenticator().get_api_base().rstrip("/")
@ -131,18 +131,18 @@ class XAIVideoConfig(BaseVideoConfig):
def validate_environment(
self,
headers: dict,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
model: str,
api_key: str | None = None,
litellm_params: GenericLiteLLMParams | None = None,
) -> dict:
) -> dict: # mutable-ok: provider JSON body and base-class dict signature
from litellm.llms.xai.oauth import (
XAIOAuthAuthenticator,
XAIOAuthError,
should_use_xai_oauth,
)
params: Final = litellm_params.model_dump() if litellm_params is not None else {}
params: Final = litellm_params.model_dump() if litellm_params is not None else {} # mutable-ok: provider JSON body and base-class dict signature
resolved_api_key: Final = api_key or (litellm_params.api_key if litellm_params else None)
dynamic_api_key: Final = XAIModelInfo.get_api_key(resolved_api_key)
if should_use_xai_oauth(params) and not dynamic_api_key:
@ -174,7 +174,7 @@ class XAIVideoConfig(BaseVideoConfig):
self,
model: str,
api_base: str | None,
litellm_params: dict,
litellm_params: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> str:
resolved: Final = self._resolve_api_base(
api_base=api_base,
@ -190,11 +190,11 @@ class XAIVideoConfig(BaseVideoConfig):
model: str,
prompt: str,
api_base: str,
video_create_optional_request_params: dict,
video_create_optional_request_params: dict, # mutable-ok: provider JSON body and base-class dict signature
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> tuple[dict, RequestFiles, str]:
copied: Final = {
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> tuple[dict, RequestFiles, str]: # mutable-ok: provider JSON body and base-class dict signature
copied: Final = { # mutable-ok: provider JSON body and base-class dict signature
key: video_create_optional_request_params[key]
for key in (
"image",
@ -207,13 +207,13 @@ class XAIVideoConfig(BaseVideoConfig):
if video_create_optional_request_params.get(key) is not None
}
return (
{
{ # mutable-ok: provider JSON body and base-class dict signature
"model": XAIModelInfo.get_base_model(model) or model,
**({"prompt": prompt} if prompt else {}),
**({"prompt": prompt} if prompt else {}), # mutable-ok: provider JSON body and base-class dict signature
**copied,
**({"duration": 6} if "duration" not in copied else {}),
**({"duration": 6} if "duration" not in copied else {}), # mutable-ok: provider JSON body and base-class dict signature
},
[],
[], # mutable-ok: provider JSON body and base-class dict signature
api_base,
)
@ -223,14 +223,14 @@ class XAIVideoConfig(BaseVideoConfig):
raw_response: httpx.Response,
logging_obj: "LiteLLMLoggingObj",
custom_llm_provider: str | None = None,
request_data: dict | None = None,
request_data: dict | None = None, # mutable-ok: provider JSON body and base-class dict signature
) -> VideoObject:
response_data: Final = raw_response.json()
request_id: Final = response_data.get("request_id") or response_data.get("id")
if not request_id:
raise ValueError(f"xAI video generation response missing request_id: {response_data}")
usage: Final = response_data.get("usage") or {}
usage: Final = response_data.get("usage") or {} # mutable-ok: provider JSON body and base-class dict signature
video_obj: Final = VideoObject(
id=str(request_id),
object="video",
@ -241,7 +241,7 @@ class XAIVideoConfig(BaseVideoConfig):
)
if custom_llm_provider:
video_obj.id = encode_video_id_with_provider(video_obj.id, custom_llm_provider, model)
video_obj.usage = usage if isinstance(usage, dict) else {}
video_obj.usage = usage if isinstance(usage, dict) else {} # mutable-ok: provider JSON body and base-class dict signature
video_obj._hidden_params["video_url"] = None
return video_obj
@ -257,9 +257,9 @@ class XAIVideoConfig(BaseVideoConfig):
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> tuple[str, dict]:
return self._video_resource_url(api_base, video_id), {}
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> tuple[str, dict]: # mutable-ok: provider JSON body and base-class dict signature
return self._video_resource_url(api_base, video_id), {} # mutable-ok: provider JSON body and base-class dict signature
def transform_video_status_retrieve_response(
self,
@ -270,7 +270,7 @@ class XAIVideoConfig(BaseVideoConfig):
response_data: Final = raw_response.json()
status_raw: Final = str(response_data.get("status") or "processing").lower()
status: Final = _STATUS_MAP.get(status_raw, status_raw)
video_meta: Final = response_data.get("video") or {}
video_meta: Final = response_data.get("video") or {} # mutable-ok: provider JSON body and base-class dict signature
video_url: Final = video_meta.get("url") if isinstance(video_meta, dict) else None
seconds: Final = (
str(video_meta.get("duration"))
@ -291,7 +291,7 @@ class XAIVideoConfig(BaseVideoConfig):
model=response_data.get("model"),
progress=response_data.get("progress"),
seconds=seconds,
usage=response_data.get("usage") if isinstance(response_data.get("usage"), dict) else {},
usage=response_data.get("usage") if isinstance(response_data.get("usage"), dict) else {}, # mutable-ok: provider JSON body and base-class dict signature
)
video_obj._hidden_params["video_url"] = video_url
if custom_llm_provider and video_obj.id and video_obj.id != "unknown":
@ -303,10 +303,10 @@ class XAIVideoConfig(BaseVideoConfig):
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
variant: str | None = None,
) -> tuple[str, dict]:
return self._video_resource_url(api_base, video_id), {}
) -> tuple[str, dict]: # mutable-ok: provider JSON body and base-class dict signature
return self._video_resource_url(api_base, video_id), {} # mutable-ok: provider JSON body and base-class dict signature
def _video_cdn_url(self, raw_response: httpx.Response) -> str | None:
content_type: Final = (raw_response.headers.get("content-type") or "").lower()
@ -315,7 +315,7 @@ class XAIVideoConfig(BaseVideoConfig):
payload: Final = raw_response.json()
if not isinstance(payload, dict):
return None
video_meta: Final = payload.get("video") or {}
video_meta: Final = payload.get("video") or {} # mutable-ok: provider JSON body and base-class dict signature
url: Final = video_meta.get("url") if isinstance(video_meta, dict) else None
if isinstance(url, str) and url:
return url
@ -355,9 +355,9 @@ class XAIVideoConfig(BaseVideoConfig):
prompt: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
extra_body: dict[str, object] | None = None,
) -> tuple[str, dict]:
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
extra_body: dict[str, object] | None = None, # mutable-ok: provider JSON body and base-class dict signature
) -> tuple[str, dict]: # mutable-ok: provider JSON body and base-class dict signature
raise NotImplementedError("Video remix is not supported by xAI Imagine API")
def transform_video_remix_response(
@ -372,12 +372,12 @@ class XAIVideoConfig(BaseVideoConfig):
self,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
after: str | None = None,
limit: int | None = None,
order: str | None = None,
extra_query: dict[str, object] | None = None,
) -> tuple[str, dict]:
extra_query: dict[str, object] | None = None, # mutable-ok: provider JSON body and base-class dict signature
) -> tuple[str, dict]: # mutable-ok: provider JSON body and base-class dict signature
raise NotImplementedError("Video listing is not supported by xAI Imagine API")
def transform_video_list_response(
@ -385,7 +385,7 @@ class XAIVideoConfig(BaseVideoConfig):
raw_response: httpx.Response,
logging_obj: "LiteLLMLoggingObj",
custom_llm_provider: str | None = None,
) -> dict[str, str]:
) -> dict[str, str]: # mutable-ok: provider JSON body and base-class dict signature
raise NotImplementedError("Video listing is not supported by xAI Imagine API")
def transform_video_delete_request(
@ -393,8 +393,8 @@ class XAIVideoConfig(BaseVideoConfig):
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> tuple[str, dict]:
headers: dict, # mutable-ok: provider JSON body and base-class dict signature
) -> tuple[str, dict]: # mutable-ok: provider JSON body and base-class dict signature
raise NotImplementedError("Video delete is not supported by xAI Imagine API")
def transform_video_delete_response(

74
litellm/proxy/image_endpoints/endpoints.py
View file

@ -54,13 +54,13 @@ async def uploadfile_to_bytesio(upload: UploadFile) -> io.BytesIO:
async def batch_to_bytesio(
uploads: Sequence[UploadFile] | None,
) -> list[io.BytesIO] | None:
) -> list[io.BytesIO] | None: # mutable-ok: provider JSON body and base-class dict signature
"""
Convert a sequence of UploadFiles to a list of BytesIO buffers, or None.
"""
if not uploads:
return None
return [await uploadfile_to_bytesio(u) for u in uploads]
return [await uploadfile_to_bytesio(u) for u in uploads] # mutable-ok: provider JSON body and base-class dict signature
def _is_image_reference_string(value: str) -> bool:
@ -95,10 +95,10 @@ async def _normalize_image_values(values: tuple[object, ...], field: str) -> obj
coerced: Final = tuple([await _coerce_image_part(value, field) for value in values])
if len(coerced) == 1 and isinstance(coerced[0], str):
return coerced[0]
return list(coerced)
return list(coerced) # mutable-ok: provider JSON body and base-class dict signature
def _json_image_values(data: dict[str, object], field: str) -> tuple[object, ...]:
def _json_image_values(data: dict[str, object], field: str) -> tuple[object, ...]: # mutable-ok: provider JSON body and base-class dict signature
if field not in data:
return ()
raw: Final = data[field]
@ -108,30 +108,30 @@ def _json_image_values(data: dict[str, object], field: str) -> tuple[object, ...
async def _normalized_image_edit_fields(
values_by_field: dict[str, tuple[object, ...]],
) -> dict[str, object]:
values_by_field: dict[str, tuple[object, ...]], # mutable-ok: provider JSON body and base-class dict signature
) -> dict[str, object]: # mutable-ok: provider JSON body and base-class dict signature
image: Final = await _normalize_image_values(values_by_field["image"], "image")
mask: Final = await _normalize_image_values(values_by_field["mask"], "mask")
return {
**({"image": image} if image is not None else {}),
**({"mask": mask} if mask is not None else {}),
return { # mutable-ok: provider JSON body and base-class dict signature
**({"image": image} if image is not None else {}), # mutable-ok: provider JSON body and base-class dict signature
**({"mask": mask} if mask is not None else {}), # mutable-ok: provider JSON body and base-class dict signature
}
async def _image_edit_assets_from_request(
request: Request,
data: dict[str, object],
) -> dict[str, object]:
data: dict[str, object], # mutable-ok: provider JSON body and base-class dict signature
) -> dict[str, object]: # mutable-ok: provider JSON body and base-class dict signature
form: Final = await request.form() if _is_form_content_type(request.headers.get("content-type", "")) else None
if form is None:
return {
**{key: value for key, value in data.items() if key not in {"image[]", "mask[]"}},
return { # mutable-ok: provider JSON body and base-class dict signature
**{key: value for key, value in data.items() if key not in {"image[]", "mask[]"}}, # mutable-ok: provider JSON body and base-class dict signature
**await _normalized_image_edit_fields(
{field: _json_image_values(data, field) for field, _alias in _IMAGE_EDIT_FILE_FIELDS}
{field: _json_image_values(data, field) for field, _alias in _IMAGE_EDIT_FILE_FIELDS} # mutable-ok: provider JSON body and base-class dict signature
),
}
form_values: Final = {
form_values: Final = { # mutable-ok: provider JSON body and base-class dict signature
name: _form_field_values(form, name) for field, alias in _IMAGE_EDIT_FILE_FIELDS for name in (field, alias)
}
conflicts: Final = tuple(
@ -142,31 +142,31 @@ async def _image_edit_assets_from_request(
status_code=422,
detail=f"Cannot specify both '{conflicts[0]}' and '{conflicts[0]}[]'",
)
return {
**{key: value for key, value in data.items() if key not in {"image[]", "mask[]"}},
return { # mutable-ok: provider JSON body and base-class dict signature
**{key: value for key, value in data.items() if key not in {"image[]", "mask[]"}}, # mutable-ok: provider JSON body and base-class dict signature
**await _normalized_image_edit_fields(
{field: form_values[field] or form_values[alias] for field, alias in _IMAGE_EDIT_FILE_FIELDS}
{field: form_values[field] or form_values[alias] for field, alias in _IMAGE_EDIT_FILE_FIELDS} # mutable-ok: provider JSON body and base-class dict signature
),
}
@router.post(
"/v1/images/generations",
dependencies=[Depends(user_api_key_auth)],
dependencies=[Depends(user_api_key_auth)], # mutable-ok: provider JSON body and base-class dict signature
response_class=ORJSONResponse,
tags=["images"],
tags=["images"], # mutable-ok: provider JSON body and base-class dict signature
)
@router.post(
"/images/generations",
dependencies=[Depends(user_api_key_auth)],
dependencies=[Depends(user_api_key_auth)], # mutable-ok: provider JSON body and base-class dict signature
response_class=ORJSONResponse,
tags=["images"],
tags=["images"], # mutable-ok: provider JSON body and base-class dict signature
)
@router.post(
"/openai/deployments/{model:path}/images/generations",
dependencies=[Depends(user_api_key_auth)],
dependencies=[Depends(user_api_key_auth)], # mutable-ok: provider JSON body and base-class dict signature
response_class=ORJSONResponse,
tags=["images"],
tags=["images"], # mutable-ok: provider JSON body and base-class dict signature
) # azure compatible endpoint
async def image_generation(
request: Request,
@ -185,7 +185,7 @@ async def image_generation(
version,
)
data = {}
data = {} # mutable-ok: provider JSON body and base-class dict signature
try:
# Use orjson to parse JSON data, orjson speeds up requests significantly
body: Final = await request.body()
@ -227,7 +227,7 @@ async def image_generation(
"role": "user",
"content": prompt_value,
}
data["messages"] = [user_message]
data["messages"] = [user_message] # mutable-ok: provider JSON body and base-class dict signature
data = await proxy_logging_obj.pre_call_hook(
user_api_key_dict=user_api_key_dict, data=data, call_type="image_generation"
)
@ -257,7 +257,7 @@ async def image_generation(
)
### RESPONSE HEADERS ###
hidden_params: Final = getattr(response, "_hidden_params", {}) or {}
hidden_params: Final = getattr(response, "_hidden_params", {}) or {} # mutable-ok: provider JSON body and base-class dict signature
model_id: Final = hidden_params.get("model_id", None) or ""
cache_key: Final = hidden_params.get("cache_key", None) or ""
api_base: Final = hidden_params.get("api_base", None) or ""
@ -284,7 +284,7 @@ async def image_generation(
data=data,
user_api_key_dict=user_api_key_dict,
response=response,
request_headers=dict(request.headers),
request_headers=dict(request.headers), # mutable-ok: provider JSON body and base-class dict signature
)
if callback_headers:
fastapi_response.headers.update(callback_headers)
@ -316,19 +316,19 @@ async def image_generation(
@router.post(
"/v1/images/edits",
dependencies=[Depends(user_api_key_auth)],
tags=["images"],
dependencies=[Depends(user_api_key_auth)], # mutable-ok: provider JSON body and base-class dict signature
tags=["images"], # mutable-ok: provider JSON body and base-class dict signature
)
@router.post(
"/images/edits",
dependencies=[Depends(user_api_key_auth)],
tags=["images"],
dependencies=[Depends(user_api_key_auth)], # mutable-ok: provider JSON body and base-class dict signature
tags=["images"], # mutable-ok: provider JSON body and base-class dict signature
)
@router.post(
"/openai/deployments/{model:path}/images/edits",
dependencies=[Depends(user_api_key_auth)],
dependencies=[Depends(user_api_key_auth)], # mutable-ok: provider JSON body and base-class dict signature
response_class=ORJSONResponse,
tags=["images"],
tags=["images"], # mutable-ok: provider JSON body and base-class dict signature
) # azure compatible endpoint
async def image_edit_api(
request: Request,
@ -364,16 +364,16 @@ async def image_edit_api(
version,
)
parsed_body: Final = dict(
parsed_body: Final = dict( # mutable-ok: provider JSON body and base-class dict signature
coerce_numeric_form_fields(
parsed_body=await _read_request_body(request=request),
numeric_fields=IMAGE_EDIT_NUMERIC_FORM_FIELDS,
)
)
with_assets: Final = await _image_edit_assets_from_request(request, parsed_body)
data: Final = {
data: Final = { # mutable-ok: provider JSON body and base-class dict signature
**with_assets,
**({} if "prompt" in with_assets else {"prompt": None}),
**({} if "prompt" in with_assets else {"prompt": None}), # mutable-ok: provider JSON body and base-class dict signature
"model": (
model or general_settings.get("image_generation_model", None) or user_model or with_assets.get("model")
),