Adolfo Reyna
71 lines
2.8 KiB
Python
71 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import Any, Dict, List
|
|
|
|
import httpx
|
|
|
|
from reyna_cli.env import load_hermes_env
|
|
|
|
|
|
class ImmichClient:
|
|
def __init__(self, base_url: str | None = None, api_key: str | None = None, timeout: float = 20.0):
|
|
load_hermes_env()
|
|
self.base_url = (base_url or os.getenv("IMMICH_BASE_URL") or os.getenv("IMMICH_URL") or "").rstrip("/")
|
|
self.api_key = api_key or os.getenv("IMMICH_API_KEY") or os.getenv("IMMICH_KEY") or ""
|
|
self.timeout = timeout
|
|
if not self.base_url:
|
|
raise RuntimeError("IMMICH_BASE_URL is not configured")
|
|
if not self.api_key:
|
|
raise RuntimeError("IMMICH_API_KEY is not configured")
|
|
self.client = httpx.Client(timeout=timeout, headers={"x-api-key": self.api_key, "Accept": "application/json"})
|
|
|
|
def _url(self, path: str) -> str:
|
|
if not path.startswith("/"):
|
|
path = "/" + path
|
|
return self.base_url + path
|
|
|
|
def _request(self, method: str, candidates: List[str], **kwargs: Any) -> Any:
|
|
errors: list[str] = []
|
|
for path in candidates:
|
|
try:
|
|
response = self.client.request(method, self._url(path), **kwargs)
|
|
if response.status_code == 404:
|
|
errors.append(f"{path}: 404")
|
|
continue
|
|
response.raise_for_status()
|
|
if not response.content:
|
|
return {}
|
|
return response.json()
|
|
except Exception as exc:
|
|
errors.append(f"{path}: {exc}")
|
|
raise RuntimeError("; ".join(errors))
|
|
|
|
def server_version(self) -> Dict[str, Any]:
|
|
return self._request("GET", ["/api/server/version", "/server/version"])
|
|
|
|
def statistics(self) -> Dict[str, Any]:
|
|
return self._request("GET", ["/api/server/statistics", "/server/statistics", "/api/server/stats", "/server/stats"])
|
|
|
|
def albums(self) -> Any:
|
|
return self._request("GET", ["/api/albums", "/albums"])
|
|
|
|
def search(self, query: str, limit: int = 10) -> Any:
|
|
payloads = [
|
|
{"query": query, "size": limit},
|
|
{"query": query, "page": 1, "size": limit},
|
|
{"smartSearch": query, "size": limit},
|
|
]
|
|
errors: list[str] = []
|
|
for path in ["/api/search/smart", "/search/smart", "/api/search/metadata", "/search/metadata"]:
|
|
for payload in payloads:
|
|
try:
|
|
response = self.client.post(self._url(path), json=payload)
|
|
if response.status_code == 404:
|
|
break
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as exc:
|
|
errors.append(f"{path}: {exc}")
|
|
raise RuntimeError("; ".join(errors))
|