fix(fal_ai): accept every FileTypes image input and derive gpt-image qualities from pricing rows

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
kerry 2026-09-20 05:32:02 +00:00
parent cf581bf327
commit cc7dce6a21
4 changed files with 78 additions and 65 deletions

View file

@ -37,36 +37,28 @@ PARAM_TRANSLATION: Final[Mapping[str, str]] = MappingProxyType(
@runtime_checkable
class _Readable(Protocol):
def read(self) -> bytes: ...
@runtime_checkable
class _Tellable(Protocol):
class _SeekableBinaryReader(Protocol):
def tell(self) -> int: ...
def seek(self, offset: int) -> int: ...
@runtime_checkable
class _Seekable(Protocol):
def seek(self, position: int) -> int: ...
def read(self) -> bytes: ...
def _read_image_bytes(image: object) -> bytes:
if isinstance(image, bytes):
return image
if isinstance(image, tuple) and len(image) >= 2:
if isinstance(image, tuple):
return _read_image_bytes(image[1])
if isinstance(image, os.PathLike):
return Path(image).read_bytes()
if isinstance(image, str) or not isinstance(image, _Readable):
raise ValueError(f"Unsupported image type for Fal AI image edit: {type(image).__name__}")
position: Final = image.tell() if isinstance(image, _Tellable) else 0
if isinstance(image, _Seekable):
if isinstance(image, _SeekableBinaryReader):
position: Final = image.tell()
image.seek(0)
data: Final = image.read()
if isinstance(image, _Seekable):
data: Final = image.read()
image.seek(position)
return data
return data
raise ValueError(f"Unsupported image type for Fal AI image edit: {type(image).__name__}")
def _to_data_url(image: object) -> str:

View file

@ -23,7 +23,6 @@ SUPPORTED_OPENAI_PARAMS: Final[tuple[OpenAIImageGenerationOptionalParams, ...]]
"response_format",
"size",
)
SUPPORTED_QUALITIES: Final[frozenset[str]] = frozenset({"auto", "low", "medium", "high"})
OPENAI_QUALITY_ALIASES: Final[Mapping[str, str]] = MappingProxyType({"hd": "high", "standard": "medium"})
@ -38,23 +37,33 @@ def map_gpt_image_size(size: object) -> object:
return image_size
def supported_gpt_image_qualities(model: str) -> frozenset[str]:
suffix: Final = f"/{model}"
keyed: Final = frozenset(
key.removeprefix("fal_ai/").split("/")[0]
for key in litellm.model_cost
if key.startswith("fal_ai/")
and key.endswith(suffix)
and key.removeprefix("fal_ai/").removesuffix(suffix).count("/") == 1
def supported_gpt_image_qualities(
model: str, model_cost: Mapping[str, Mapping[str, object]] | None = None
) -> frozenset[str]:
costs: Final = litellm.model_cost if model_cost is None else model_cost
endpoint: Final[str] = model.removeprefix("fal_ai/")
qualified_endpoint: Final[str] = endpoint if endpoint.startswith("openai/") else f"openai/{endpoint}"
qualities: Final[frozenset[str]] = frozenset(
parts[1]
for key in costs
if (parts := key.split("/"))[0] == "fal_ai"
and len(parts) > 3
and "-x-" in parts[2]
and "/".join(parts[3:]) == qualified_endpoint
)
return keyed | frozenset(("auto",)) if keyed else SUPPORTED_QUALITIES
return qualities | {"auto"} if qualities else frozenset()
def map_gpt_image_quality(quality: object, model: str) -> object:
def map_gpt_image_quality(
quality: object, model: str, model_cost: Mapping[str, Mapping[str, object]] | None = None
) -> object:
if not isinstance(quality, str):
return quality
normalized: Final[str] = OPENAI_QUALITY_ALIASES.get(quality, quality)
return normalized if normalized in supported_gpt_image_qualities(model) else "auto"
supported: Final[frozenset[str]] = supported_gpt_image_qualities(model, model_cost)
if not supported:
return normalized
return normalized if normalized in supported else "auto"
class FalAIGPTImage2Config(FalAIBaseConfig):

View file

@ -1,8 +1,8 @@
import base64
import io
import json
import tempfile
from pathlib import Path
from typing import Final
import httpx
import pytest
@ -16,20 +16,6 @@ from litellm.utils import ProviderConfigManager
PNG_BYTES = b"\x89PNG\r\n\x1a\n" + b"\x00" * 16
class GenericFileLike:
def __init__(self, data: bytes):
self._buffer = io.BytesIO(data)
def read(self) -> bytes:
return self._buffer.read()
def seek(self, position: int) -> int:
return self._buffer.seek(position)
def tell(self) -> int:
return self._buffer.tell()
def test_fal_ai_resolves_to_image_edit_config():
config = ProviderConfigManager.get_provider_image_edit_config(
model="openai/gpt-image-2.5/flare/edit", provider=LlmProviders.FAL_AI
@ -101,27 +87,36 @@ def test_transform_request_inlines_local_images_as_data_urls_and_keeps_remote_ur
assert "mask" not in body
@pytest.mark.parametrize("input_kind", ("path", "tuple_bytes", "tuple_file_like", "file_like"))
def test_transform_request_accepts_openai_file_types(tmp_path, input_kind):
image_path: Final[Path] = tmp_path / "in.png"
image_path.write_bytes(PNG_BYTES)
image: Final[object] = {
"path": image_path,
"tuple_bytes": ("in.png", PNG_BYTES),
"tuple_file_like": ("in.png", io.BytesIO(PNG_BYTES), "image/png"),
"file_like": GenericFileLike(PNG_BYTES),
}[input_kind]
expected_data_url: Final = "data:image/png;base64," + base64.b64encode(PNG_BYTES).decode()
body, files = FalAIImageEditConfig().transform_image_edit_request(
model="openai/gpt-image-2",
@pytest.mark.parametrize(
"image_factory",
[
pytest.param(lambda path: ("red.png", PNG_BYTES), id="filename-bytes-tuple"),
pytest.param(lambda path: ("red.png", PNG_BYTES, "image/png"), id="three-tuple-with-content-type"),
pytest.param(lambda path: path, id="path"),
pytest.param(lambda path: io.FileIO(str(path), "rb"), id="file-io"),
pytest.param(
lambda path: tempfile.SpooledTemporaryFile(suffix=".png"),
id="spooled-temp-file",
),
],
)
def test_transform_request_reads_every_file_types_input(tmp_path, image_factory):
path = Path(tmp_path) / "red.png"
path.write_bytes(PNG_BYTES)
image = image_factory(path)
if isinstance(image, tempfile.SpooledTemporaryFile):
image.write(PNG_BYTES)
image.seek(3)
body, _ = FalAIImageEditConfig().transform_image_edit_request(
model="openai/gpt-image-2.5/flare/edit",
prompt="make it blue",
image=image,
image_edit_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert files == ()
assert body["image_urls"] == (expected_data_url,)
expected_data_url = "data:image/png;base64," + base64.b64encode(PNG_BYTES).decode()
assert body["image_urls"][0] == expected_data_url
def test_transform_response_maps_fal_images():

View file

@ -7,7 +7,10 @@ from litellm.llms.fal_ai.image_generation import (
FalAINanoBananaConfig,
get_fal_ai_image_generation_config,
)
from litellm.llms.fal_ai.image_generation.gpt_image_2_transformation import map_gpt_image_quality
from litellm.llms.fal_ai.image_generation.gpt_image_2_transformation import (
map_gpt_image_quality,
supported_gpt_image_qualities,
)
from litellm.types.utils import ImageObject, ImageResponse
@ -163,8 +166,22 @@ def test_map_openai_params_quality_tiers_follow_model(model, quality, expected):
) == {"quality": expected}
def test_map_gpt_image_quality_derives_supported_tiers_from_pricing_metadata():
assert map_gpt_image_quality("xhigh", "openai/gpt-image-2.5/flare/text-to-image") == "xhigh"
assert map_gpt_image_quality("xhigh", "openai/gpt-image-2") == "auto"
assert map_gpt_image_quality("xhigh", "openai/unknown-model") == "auto"
assert map_gpt_image_quality("high", "openai/unknown-model") == "high"
@pytest.mark.parametrize(
"model",
[
"some-new-model",
"openai/some-new-model",
"fal_ai/openai/some-new-model",
],
)
def test_supported_qualities_derived_from_pricing_rows(model):
model_cost = {
"fal_ai/xhigh/1024-x-1024/openai/some-new-model": {},
"fal_ai/low/1024-x-1024/openai/some-new-model": {},
"fal_ai/max/1024-x-1024/openai/other-model": {},
}
assert supported_gpt_image_qualities(model, model_cost) == {"xhigh", "low", "auto"}
def test_map_gpt_image_quality_passes_through_when_no_pricing_rows():
assert map_gpt_image_quality("xhigh", "some-new-model", {}) == "xhigh"