Trakt-Rater/trakt.py

import asyncio
import httpx


class TraktClient:
    BASE_URL = "https://api.trakt.tv"

    def __init__(self, client_id: str, access_token: str):
        self.headers = {
            "Content-Type": "application/json",
            "trakt-api-version": "2",
            "trakt-api-key": client_id,
            "Authorization": f"Bearer {access_token}",
        }

    async def get_watched_and_ratings(self):
        async with httpx.AsyncClient(timeout=60) as client:
            watched_r, ratings_r = await asyncio.gather(
                client.get(f"{self.BASE_URL}/sync/watched/movies", headers=self.headers),
                client.get(f"{self.BASE_URL}/sync/ratings/movies", headers=self.headers),
            )
        watched_r.raise_for_status()
        ratings_r.raise_for_status()
        return watched_r.json(), ratings_r.json()

    async def remove_rating(self, trakt_id: int):
        async with httpx.AsyncClient(timeout=30) as client:
            r = await client.post(
                f"{self.BASE_URL}/sync/ratings/remove",
                headers=self.headers,
                json={"movies": [{"ids": {"trakt": trakt_id}}]},
            )
        r.raise_for_status()
        return r.json()

    async def rate_movie(self, trakt_id: int, rating: int):
        async with httpx.AsyncClient(timeout=30) as client:
            r = await client.post(
                f"{self.BASE_URL}/sync/ratings",
                headers=self.headers,
                json={"movies": [{"rating": rating, "ids": {"trakt": trakt_id}}]},
            )
        r.raise_for_status()
        return r.json()