bruno
2f2a5e9b4a
Introduce an admin portal (React + Nginx), WebSocket routing, and API versioning middleware with `/api/v1/` prefix deprecation. Add master API key authentication, new Prometheus metrics for AI token consumption and active WebSockets, and extend S3 config with a public endpoint URL. Update test paths and fixtures to align with the new routing structure.
64 lines
2.7 KiB
Python
64 lines
2.7 KiB
Python
"""
|
|
Client module for interacting with Imago API resources.
|
|
"""
|
|
from typing import Any, Dict, Optional, List
|
|
import io
|
|
|
|
from httpx import AsyncClient, Response
|
|
|
|
from ..models import HubImage
|
|
from ..websocket import PipelineStream
|
|
from ..utils import with_retry
|
|
|
|
class ImagesResource:
|
|
"""Resource defining image operations in the Imago API."""
|
|
def __init__(self, client: AsyncClient):
|
|
self._client = client
|
|
self.ws_url = f"{self._client.base_url}"
|
|
|
|
async def _handle_request(self, method: str, path: str, **kwargs) -> Any:
|
|
async def _req():
|
|
resp = await self._client.request(method, path, **kwargs)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
return await with_retry(_req)
|
|
|
|
async def list(self, page: int = 1, size: int = 50, tags: Optional[str] = None) -> Dict[str, Any]:
|
|
params: dict = {"page": page, "size": size}
|
|
if tags is not None:
|
|
params["tags"] = tags
|
|
return await self._handle_request("GET", "/api/v1/images", params=params)
|
|
|
|
async def get(self, image_id: int) -> HubImage:
|
|
data = await self._handle_request("GET", f"/api/v1/images/{image_id}")
|
|
return HubImage.from_dict(data)
|
|
|
|
async def get_status(self, image_id: int) -> Dict[str, Any]:
|
|
return await self._handle_request("GET", f"/api/v1/images/{image_id}/status")
|
|
|
|
async def upload(self, file_content: bytes, filename: str, content_type: str = "image/jpeg") -> HubImage:
|
|
files = {"file": (filename, file_content, content_type)}
|
|
data = await self._handle_request("POST", "/api/v1/images/upload", files=files)
|
|
return HubImage.from_dict(data)
|
|
|
|
async def delete(self, image_id: int) -> Dict[str, Any]:
|
|
return await self._handle_request("DELETE", f"/api/v1/images/{image_id}")
|
|
|
|
async def get_all_tags(self) -> List[str]:
|
|
return await self._handle_request("GET", "/api/v1/images/tags/all")
|
|
|
|
async def reprocess(self, image_id: int) -> Dict[str, Any]:
|
|
return await self._handle_request("POST", f"/api/v1/images/{image_id}/reprocess")
|
|
|
|
async def get_download_url(self, image_id: int) -> str:
|
|
data = await self._handle_request("GET", f"/api/v1/images/{image_id}/download-url")
|
|
return data.get("download_url")
|
|
|
|
async def get_thumbnail_url(self, image_id: int) -> str:
|
|
data = await self._handle_request("GET", f"/api/v1/images/{image_id}/thumbnail-url")
|
|
return data.get("thumbnail_url")
|
|
|
|
def stream_pipeline(self, image_id: int, api_key: str) -> PipelineStream:
|
|
"""Returns a PipelineStream object for tracking the pipeline status in real-time."""
|
|
return PipelineStream(self.ws_url, api_key, image_id)
|