client.py•3.15 kB
import os
import requests
LIDARR_URL = os.getenv("LIDARR_URL")
LIDARR_API_KEY = os.getenv("LIDARR_API_KEY")
if not (LIDARR_URL and LIDARR_API_KEY):
# On laisse le warning mais le client peut quand même être importé pour tests
print("Warning: LIDARR_URL or LIDARR_API_KEY not set. Configure them in .env")
SESSION = requests.Session()
SESSION.headers.update({"X-Api-Key": LIDARR_API_KEY, "Accept": "application/json"})
def _url(path: str) -> str:
base = LIDARR_URL.rstrip("/")
return f"{base}{path}"
# Artists
def list_artists() -> list:
resp = SESSION.get(_url("/api/v1/artist"))
resp.raise_for_status()
return resp.json()
def get_artist(artist_id: int) -> dict:
resp = SESSION.get(_url(f"/api/v1/artist/{artist_id}"))
resp.raise_for_status()
return resp.json()
def create_artist(payload: dict) -> dict:
resp = SESSION.post(_url("/api/v1/artist"), json=payload)
resp.raise_for_status()
return resp.json()
def update_artist(artist_id: int, payload: dict) -> dict:
resp = SESSION.put(_url(f"/api/v1/artist/{artist_id}"), json=payload)
resp.raise_for_status()
return resp.json()
def delete_artist(artist_id: int) -> dict:
resp = SESSION.delete(_url(f"/api/v1/artist/{artist_id}"))
resp.raise_for_status()
return resp.json()
# Albums / Releases
def list_albums_by_artist(artist_id: int) -> list:
# Try common endpoints; adjust depending on Lidarr version
resp = SESSION.get(_url(f"/api/v1/release?artistId={artist_id}"))
if resp.status_code == 404:
resp = SESSION.get(_url(f"/api/v1/artist/{artist_id}/releases"))
resp.raise_for_status()
return resp.json()
def get_album(album_id: int) -> dict:
resp = SESSION.get(_url(f"/api/v1/release/{album_id}"))
resp.raise_for_status()
return resp.json()
def create_album(payload: dict) -> dict:
resp = SESSION.post(_url("/api/v1/release"), json=payload)
resp.raise_for_status()
return resp.json()
def update_album(album_id: int, payload: dict) -> dict:
resp = SESSION.put(_url(f"/api/v1/release/{album_id}"), json=payload)
resp.raise_for_status()
return resp.json()
def delete_album(album_id: int) -> dict:
resp = SESSION.delete(_url(f"/api/v1/release/{album_id}"))
resp.raise_for_status()
return resp.json()
# Tracks
def list_tracks_by_album(album_id: int) -> list:
resp = SESSION.get(_url(f"/api/v1/release/{album_id}/tracks"))
resp.raise_for_status()
return resp.json()
def get_track(track_id: int) -> dict:
resp = SESSION.get(_url(f"/api/v1/track/{track_id}"))
resp.raise_for_status()
return resp.json()
def create_track(payload: dict) -> dict:
resp = SESSION.post(_url("/api/v1/track"), json=payload)
resp.raise_for_status()
return resp.json()
def update_track(track_id: int, payload: dict) -> dict:
resp = SESSION.put(_url(f"/api/v1/track/{track_id}"), json=payload)
resp.raise_for_status()
return resp.json()
def delete_track(track_id: int) -> dict:
resp = SESSION.delete(_url(f"/api/v1/track/{track_id}"))
resp.raise_for_status()
return resp.json()