(uri: str)
| 219 | |
| 220 | |
| 221 | async def _load_image_bytes_from_uri(uri: str) -> bytes: |
| 222 | if not uri: |
| 223 | raise HTTPException(status_code=400, detail="Image URI cannot be empty") |
| 224 | |
| 225 | if uri.startswith("data:image"): |
| 226 | _, image_bytes = _decode_data_url(uri) |
| 227 | return image_bytes |
| 228 | |
| 229 | if uri.startswith("http://") or uri.startswith("https://") or "/tmp/" in uri: |
| 230 | image_bytes = await retrieve_image_data(uri) |
| 231 | if image_bytes: |
| 232 | return image_bytes |
| 233 | raise HTTPException(status_code=400, detail=f"Failed to load image from {uri}") |
| 234 | |
| 235 | raise HTTPException(status_code=400, detail=f"Unsupported image URI: {uri}") |
| 236 | |
| 237 | |
| 238 | def _coerce_gemini_contents(raw_contents: Optional[List[Any]]) -> List[GeminiContent]: |
no test coverage detected