(image_uri)
| 40 | return image.format |
| 41 | |
| 42 | async def get_image_base64_string(image_uri): |
| 43 | |
| 44 | # Image in local file system |
| 45 | if image_url_is_on_localhost(image_uri): |
| 46 | local_file_path = CONFIG.PATH_TO_VOLUME + "/imgs/" + image_uri.split("/imgs/")[1] |
| 47 | |
| 48 | with open(local_file_path, "rb") as image_file: |
| 49 | image_bytes = image_file.read() |
| 50 | base64_string = base64.b64encode(image_bytes).decode("utf-8") |
| 51 | return base64_string |
| 52 | |
| 53 | # Normal url |
| 54 | async with aiohttp.ClientSession() as session: |
| 55 | async with session.get(url=image_uri, proxy=CONFIG.PROXY) as response: |
| 56 | if response.status == 200: |
| 57 | image_bytes = await response.read() |
| 58 | # Encode the bytes to a base64 string |
| 59 | base64_string = base64.b64encode(image_bytes).decode("utf-8") |
| 60 | return base64_string |
| 61 | else: |
| 62 | raise_provider_api_error(f"Failed to fetch image from {image_uri}") |
no test coverage detected