diff --git a/litellm/llms/xai/image_edit/__init__.py b/litellm/llms/xai/image_edit/__init__.py index 5cae7020a92..79664e14d6b 100644 --- a/litellm/llms/xai/image_edit/__init__.py +++ b/litellm/llms/xai/image_edit/__init__.py @@ -1,3 +1,3 @@ from .transformation import XAIImageEditConfig -__all__ = ["XAIImageEditConfig"] +__all__ = ["XAIImageEditConfig"] # mutable-ok: provider JSON body and base-class dict signature diff --git a/litellm/llms/xai/image_edit/transformation.py b/litellm/llms/xai/image_edit/transformation.py index 15762d76391..a109f1051b2 100644 --- a/litellm/llms/xai/image_edit/transformation.py +++ b/litellm/llms/xai/image_edit/transformation.py @@ -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): diff --git a/litellm/llms/xai/image_generation/__init__.py b/litellm/llms/xai/image_generation/__init__.py index cc249fdf9ff..c1b24b609c3 100644 --- a/litellm/llms/xai/image_generation/__init__.py +++ b/litellm/llms/xai/image_generation/__init__.py @@ -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: diff --git a/litellm/llms/xai/image_generation/transformation.py b/litellm/llms/xai/image_generation/transformation.py index 327f7891ae2..2db407eb450 100644 --- a/litellm/llms/xai/image_generation/transformation.py +++ b/litellm/llms/xai/image_generation/transformation.py @@ -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 diff --git a/litellm/llms/xai/videos/__init__.py b/litellm/llms/xai/videos/__init__.py index 1e9330ffba1..45650657059 100644 --- a/litellm/llms/xai/videos/__init__.py +++ b/litellm/llms/xai/videos/__init__.py @@ -1,3 +1,3 @@ from .transformation import XAIVideoConfig -__all__ = ["XAIVideoConfig"] +__all__ = ["XAIVideoConfig"] # mutable-ok: provider JSON body and base-class dict signature diff --git a/litellm/llms/xai/videos/transformation.py b/litellm/llms/xai/videos/transformation.py index 2755fe6660d..039dce3a9fa 100644 --- a/litellm/llms/xai/videos/transformation.py +++ b/litellm/llms/xai/videos/transformation.py @@ -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( diff --git a/litellm/proxy/image_endpoints/endpoints.py b/litellm/proxy/image_endpoints/endpoints.py index a892cc1ab46..615e821270d 100644 --- a/litellm/proxy/image_endpoints/endpoints.py +++ b/litellm/proxy/image_endpoints/endpoints.py @@ -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") ),