Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/multimodal/fetcher.py: 40%

1"""Async image fetcher that converts URLs to base64-encoded image parts.""" 

2 

3from __future__ import annotations 

4 

5import base64 

6 

7import httpx 

8 

9from lexigram.ai.llm.exceptions import LLMError 

10from lexigram.contracts.ai.multimodal import ImageBase64Part 

11from lexigram.logging import ( 

12 get_logger, 

13) 

14 

15logger = get_logger(__name__) 

16 

17 

18async def fetch_image_as_base64(url: str, timeout: float = 10.0) -> ImageBase64Part: 

19 """Fetch an image URL and return it as a base64-encoded part. 

20 

21 Used by clients that cannot pass image URLs through to the provider 

22 (e.g. Ollama, AWS Bedrock). The fetch is done once per message 

23 construction — callers should cache results when sending the same 

24 image in multiple turns. 

25 

26 Args: 

27 url: Public HTTP/HTTPS URL of the image. 

28 timeout: Request timeout in seconds (default: 10). 

29 

30 Returns: 

31 :class:`~lexigram.contracts.ai.multimodal.ImageBase64Part` with 

32 base64-encoded data and detected media type. 

33 

34 Raises: 

35 LLMError: If the HTTP request fails or times out. 

36 """ 

37 try: 

38 async with httpx.AsyncClient(timeout=timeout) as client: 

39 response = await client.get(url) 

40 response.raise_for_status() 

41 except httpx.HTTPError as exc: 

42 raise LLMError(f"Failed to fetch image from {url!r}: {exc}") from exc 

43 

44 raw = response.content 

45 media_type = ( 

46 response.headers.get("content-type", "image/jpeg").split(";")[0].strip() 

47 ) 

48 if not media_type.startswith("image/"): 

49 media_type = "image/jpeg" 

50 

51 logger.debug("image_fetched", url=url, media_type=media_type, size_bytes=len(raw)) 

52 return ImageBase64Part( 

53 data=base64.b64encode(raw).decode(), 

54 media_type=media_type, 

55 )