Skip to content

Async Client

AsyncAniList

AsyncAniList(
    api_url: str = "https://graphql.anilist.co",
    *,
    client: AsyncClient | None = None,
)

AniList API client.

Parameters:

Name Type Description Default
api_url str

The URL of the AniList API.

'https://graphql.anilist.co'
client AsyncClient | None

An httpx.AsyncClient instance used to make requests to AniList.

None
Source code in src/pyanilist/_aclient.py
def __init__(self, api_url: str = "https://graphql.anilist.co", *, client: AsyncClient | None = None) -> None:
    """
    AniList API client.

    Parameters
    ----------
    api_url : str, optional
        The URL of the AniList API.
    client : AsyncClient | None, optional
        An [`httpx.AsyncClient`](https://www.python-httpx.org/api/#asyncclient) instance
        used to make requests to AniList.

    """
    self._api_url = api_url
    self._client = (
        AsyncClient(headers={"Referer": "https://anilist.co", "User-Agent": f"pyanilist/{__version__}"})
        if client is None
        else client
    )

close async

close() -> None

Close the underlying HTTP connection.

Source code in src/pyanilist/_aclient.py
async def close(self) -> None:
    """Close the underlying HTTP connection."""
    await self._client.aclose()

get_airing_schedule async

get_airing_schedule(
    media: MediaID, *, not_yet_aired: bool | None = None
) -> AsyncIterator[AiringSchedule]

Retrieve the airing schedule for a given Media object or ID.

Parameters:

Name Type Description Default
media MediaID

The media to get the airing schedule for. Can be an ID (int), a URL (str), or a Media object.

required
not_yet_aired bool | None

Filter results to include only episodes that have not yet aired (True), exclude unaired episodes (False), or include all episodes (None).

None

Yields:

Type Description
AiringSchedule

An object representing the retrieved airing schedule.

Raises:

Type Description
MediaNotFoundError

If the provided media ID or URL does not correspond to any existing media on AniList.

RateLimitError

If the API rate limit is exceeded. The error contains information on how long to wait before retrying.

AnilistError

If any other error occurs during the API request.

Source code in src/pyanilist/_aclient.py
async def get_airing_schedule(
    self,
    media: MediaID,
    *,
    not_yet_aired: bool | None = None,
) -> AsyncIterator[AiringSchedule]:
    """
    Retrieve the airing schedule for a given `Media` object or ID.

    Parameters
    ----------
    media : MediaID
        The media to get the airing schedule for. Can be an ID (`int`), a URL (`str`), or a `Media` object.
    not_yet_aired : bool | None, optional
        Filter results to include only episodes that have not yet aired (`True`),
        exclude unaired episodes (`False`), or include all episodes (`None`).

    Yields
    ------
    AiringSchedule
        An object representing the retrieved airing schedule.

    Raises
    ------
    MediaNotFoundError
        If the provided media ID or URL does not correspond to any existing media on AniList.
    RateLimitError
        If the API rate limit is exceeded. The error contains information on how long to wait before retrying.
    AnilistError
        If any other error occurs during the API request.

    """
    variables: dict[str, Any] = {"mediaId": resolve_media_id(media)}

    if not_yet_aired is not None:
        variables["notYetAired"] = not_yet_aired

    response = await self._post(query=AIRING_SCHEDULE_QUERY, variables=variables)
    nodes = response["Media"]["airingSchedule"]["nodes"]

    for node in nodes:
        if schedule := normalize_anilist_data(node):
            yield msgspec.convert(schedule, type=AiringSchedule, strict=False)

get_characters async

get_characters(
    media: MediaID,
    *,
    sort: SortType[CharacterSort] | None = None,
    role: CharacterRole | None = None,
) -> AsyncIterator[Character]

Retrieve characters associated with a given Media object or ID.

Parameters:

Name Type Description Default
media MediaID

The media to get characters for. Can be an ID (int), a URL (str), or a Media object.

required
sort SortType[CharacterSort]

Sorting criteria for the characters. Can be an instance of CharacterSort, an iterable of CharacterSort, or None.

None
role CharacterRole | None

Filter characters by their role in the media. If None, no filtering is applied.

None

Yields:

Type Description
Character

An object representing the retrieved character.

Raises:

Type Description
MediaNotFoundError

If the provided media ID or URL does not correspond to any existing media on AniList.

RateLimitError

If the API rate limit is exceeded. The error contains information on how long to wait before retrying.

AnilistError

If any other error occurs during the API request.

Source code in src/pyanilist/_aclient.py
async def get_characters(
    self,
    media: MediaID,
    *,
    sort: SortType[CharacterSort] | None = None,
    role: CharacterRole | None = None,
) -> AsyncIterator[Character]:
    """
    Retrieve characters associated with a given `Media` object or ID.

    Parameters
    ----------
    media : MediaID
        The media to get characters for. Can be an ID (`int`), a URL (`str`), or a `Media` object.
    sort : SortType[CharacterSort], optional
        Sorting criteria for the characters.
        Can be an instance of `CharacterSort`, an iterable of `CharacterSort`, or None.
    role : CharacterRole | None, optional
        Filter characters by their role in the media. If `None`, no filtering is applied.

    Yields
    ------
    Character
        An object representing the retrieved character.

    Raises
    ------
    MediaNotFoundError
        If the provided media ID or URL does not correspond to any existing media on AniList.
    RateLimitError
        If the API rate limit is exceeded. The error contains information on how long to wait before retrying.
    AnilistError
        If any other error occurs during the API request.

    """
    variables: dict[str, Any] = {"mediaId": resolve_media_id(media)}

    if sort_key := get_sort_key(sort, CharacterSort):
        variables["sort"] = sort_key

    if role is not None:
        variables["role"] = role

    response = await self._post(query=CHARACTERS_QUERY, variables=variables)
    edges = response["Media"]["characters"]["edges"]

    for edge in edges:
        if character := edge["node"]:
            character["role"] = edge["role"]
            character["voiceActors"] = edge["voiceActors"]
            if character := normalize_anilist_data(character):  # AniList can return a `null` character.
                # This check is necessary because in some cases,
                # we may end up with an empty dictionary after normalizing.
                # See: https://github.com/Ravencentric/pyanilist/issues/29
                yield msgspec.convert(character, type=Character, strict=False)

get_media async

get_media(
    search: str | None = None,
    **kwargs: Unpack[MediaQueryParams],
) -> Media

Retrieve a single media object from AniList based on the provided query parameters.

Parameters:

Name Type Description Default
search str | None

Search term to include in the query.

None
**kwargs Unpack[MediaQueryParams]

Additional query parameters. See MediaQueryParams for the full list of supported parameters and their descriptions.

{}

Raises:

Type Description
MediaNotFoundError

If no media matches the provided query parameters on AniList.

InvalidMediaQueryError

If the query parameters are missing, empty, or contain unexpected keys.

RateLimitError

If the API rate limit is exceeded. The error contains information on how long to wait before retrying.

AnilistError

If any other error occurs during the API request.

Returns:

Type Description
Media

An object representing the retrieved media.

Source code in src/pyanilist/_aclient.py
async def get_media(self, search: str | None = None, **kwargs: Unpack[MediaQueryParams]) -> Media:
    """
    Retrieve a single media object from AniList based on the provided query parameters.

    Parameters
    ----------
    search : str | None, optional
        Search term to include in the query.
    **kwargs : Unpack[MediaQueryParams], optional
        Additional query parameters. See [`MediaQueryParams`][pyanilist.MediaQueryParams]
        for the full list of supported parameters and their descriptions.

    Raises
    ------
    MediaNotFoundError
        If no media matches the provided query parameters on AniList.
    InvalidMediaQueryError
        If the query parameters are missing, empty, or contain unexpected keys.
    RateLimitError
        If the API rate limit is exceeded. The error contains information on how long to wait before retrying.
    AnilistError
        If any other error occurs during the API request.

    Returns
    -------
    Media
        An object representing the retrieved media.

    """
    response = await self._post(
        query=MEDIA_QUERY,
        variables=to_anilist_vars(search, kwargs),
    )
    media = normalize_anilist_data(response["Media"])
    return msgspec.convert(media, type=Media, strict=False)

get_media_many async

get_media_many(
    search: str | None = None,
    **kwargs: Unpack[MediaQueryParams],
) -> AsyncIterator[Media]

Retrieve all matching media from AniList as an asynchronous iterator based on the provided query parameters.

Unlike AsyncAniList.get_media, this method does not raise a MediaNotFoundError if no results are found; the iterator will simply be empty.

Parameters:

Name Type Description Default
search str | None

Search term to include in the query.

None
**kwargs Unpack[MediaQueryParams]

Additional query parameters. See MediaQueryParams for the full list of supported parameters and their descriptions.

{}

Raises:

Type Description
InvalidMediaQueryError

If the query parameters are missing, empty, or contain unexpected keys.

RateLimitError

If the API rate limit is exceeded. The error contains information on how long to wait before retrying.

AnilistError

If any other error occurs during the API request.

Yields:

Type Description
Media

Each matching media object retrieved from AniList.

Source code in src/pyanilist/_aclient.py
async def get_media_many(
    self, search: str | None = None, **kwargs: Unpack[MediaQueryParams]
) -> AsyncIterator[Media]:
    """
    Retrieve all matching media from AniList as an asynchronous iterator
    based on the provided query parameters.

    Unlike [`AsyncAniList.get_media`][pyanilist.AsyncAniList.get_media],
    this method does not raise a [`MediaNotFoundError`][pyanilist.MediaNotFoundError]
    if no results are found; the iterator will simply be empty.

    Parameters
    ----------
    search : str | None, optional
        Search term to include in the query.
    **kwargs : Unpack[MediaQueryParams], optional
        Additional query parameters. See [`MediaQueryParams`][pyanilist.MediaQueryParams]
        for the full list of supported parameters and their descriptions.

    Raises
    ------
    InvalidMediaQueryError
        If the query parameters are missing, empty, or contain unexpected keys.
    RateLimitError
        If the API rate limit is exceeded. The error contains information on how long to wait before retrying.
    AnilistError
        If any other error occurs during the API request.

    Yields
    ------
    Media
        Each matching media object retrieved from AniList.

    """

    variables = to_anilist_vars(search, kwargs)

    # Fetch four pages in one request with 50 results per page (AniList caps at 50).
    variables["page1"] = 1
    variables["page2"] = 2
    variables["page3"] = 3
    variables["page4"] = 4
    variables["perPage"] = 50
    has_next_page: bool = True

    while has_next_page:
        response = await self._post(
            query=ALL_MEDIA_QUERY,
            variables=variables,
        )

        # As per AniList's documentation:
        # "You should only rely on hasNextPage for any pagination logic."
        # Reference: https://docs.anilist.co/guide/graphql/pagination#pageinfo
        #
        # In this case, we always grab 4 pages at a time,
        # so we only need to check if there are any more pages
        # after the last page (page4).
        has_next_page = response["page4"]["pageInfo"]["hasNextPage"]

        if has_next_page:
            # Get the next set of 4 pages.
            variables["page1"] += 4  # 1 + 4 => 5
            variables["page2"] += 4  # 2 + 4 => 6
            variables["page3"] += 4  # 3 + 4 => 7
            variables["page4"] += 4  # 4 + 4 => 8

        page1 = response["page1"]["media"]
        page2 = response["page2"]["media"]
        page3 = response["page3"]["media"]
        page4 = response["page4"]["media"]

        for media in itertools.chain(page1, page2, page3, page4):
            if node := normalize_anilist_data(media):
                # This check is necessary because in some cases,
                # we may end up with an empty dictionary after normalizing.
                # See: https://github.com/Ravencentric/pyanilist/issues/29
                yield msgspec.convert(node, type=Media, strict=False)

get_recommendations async

get_recommendations(
    media: MediaID,
    *,
    sort: SortType[RecommendationSort] | None = None,
) -> AsyncIterator[RecommendedMedia]

Retrieve recommended media based on a given Media object or ID.

Parameters:

Name Type Description Default
media MediaID

The media to get recommendations for. Can be an ID (int), a URL (str), or a Media object.

required
sort SortType[RecommendationSort]

Sorting criteria for the recommendations. Can be an instance of RecommendationSort, an iterable of RecommendationSort, or None.

None

Yields:

Type Description
RecommendedMedia

An object representing the retrieved recommended media.

Raises:

Type Description
MediaNotFoundError

If the provided media ID or URL does not correspond to any existing media on AniList.

RateLimitError

If the API rate limit is exceeded. The error contains information on how long to wait before retrying.

AnilistError

If any other error occurs during the API request.

Source code in src/pyanilist/_aclient.py
async def get_recommendations(
    self,
    media: MediaID,
    *,
    sort: SortType[RecommendationSort] | None = None,
) -> AsyncIterator[RecommendedMedia]:
    """
    Retrieve recommended media based on a given `Media` object or ID.

    Parameters
    ----------
    media : MediaID
        The media to get recommendations for. Can be an ID (`int`), a URL (`str`), or a `Media` object.
    sort : SortType[RecommendationSort], optional
        Sorting criteria for the recommendations.
        Can be an instance of `RecommendationSort`, an iterable of `RecommendationSort`, or None.

    Yields
    ------
    RecommendedMedia
        An object representing the retrieved recommended media.

    Raises
    ------
    MediaNotFoundError
        If the provided media ID or URL does not correspond to any existing media on AniList.
    RateLimitError
        If the API rate limit is exceeded. The error contains information on how long to wait before retrying.
    AnilistError
        If any other error occurs during the API request.

    """
    variables: dict[str, Any] = {"mediaId": resolve_media_id(media)}

    if sort_key := get_sort_key(sort, RecommendationSort):
        variables["sort"] = sort_key

    response = await self._post(query=RECOMMENDATIONS_QUERY, variables=variables)
    nodes = response["Media"]["recommendations"]["nodes"]

    for node in nodes:
        if rec := node["mediaRecommendation"]:  # AniList can return a `null` recommendation.
            rec["rating"] = node["rating"]
            if rec := normalize_anilist_data(rec):
                # This check is necessary because in some cases,
                # we may end up with an empty dictionary after normalizing.
                # See: https://github.com/Ravencentric/pyanilist/issues/29
                yield msgspec.convert(rec, type=RecommendedMedia, strict=False)

get_relations async

get_relations(
    media: MediaID,
) -> AsyncIterator[RelatedMedia]

Retrieve related media based on a given Media object or ID.

Parameters:

Name Type Description Default
media MediaID

The media to get related media for. Can be an ID (int), a URL (str), or a Media object.

required

Yields:

Type Description
RelatedMedia

An object representing the retrieved related media.

Raises:

Type Description
MediaNotFoundError

If the provided media ID or URL does not correspond to any existing media on AniList.

RateLimitError

If the API rate limit is exceeded. The error contains information on how long to wait before retrying.

AnilistError

If any other error occurs during the API request.

Source code in src/pyanilist/_aclient.py
async def get_relations(self, media: MediaID) -> AsyncIterator[RelatedMedia]:
    """
    Retrieve related media based on a given `Media` object or ID.

    Parameters
    ----------
    media : MediaID
        The media to get related media for. Can be an ID (`int`), a URL (`str`), or a `Media` object.

    Yields
    ------
    RelatedMedia
        An object representing the retrieved related media.

    Raises
    ------
    MediaNotFoundError
        If the provided media ID or URL does not correspond to any existing media on AniList.
    RateLimitError
        If the API rate limit is exceeded. The error contains information on how long to wait before retrying.
    AnilistError
        If any other error occurs during the API request.

    """
    response = await self._post(query=RELATIONS_QUERY, variables={"mediaId": resolve_media_id(media)})
    edges = response["Media"]["relations"]["edges"]

    for edge in edges:
        if relation := edge["node"]:  # AniList can return a `null` relation.
            relation["relationType"] = edge["relationType"]
            if relation := normalize_anilist_data(relation):
                # This check is necessary because in some cases,
                # we may end up with an empty dictionary after normalizing.
                # See: https://github.com/Ravencentric/pyanilist/issues/29
                yield msgspec.convert(relation, type=RelatedMedia, strict=False)

get_staffs async

get_staffs(
    media: MediaID,
    *,
    sort: SortType[StaffSort] | None = None,
) -> AsyncIterator[Staff]

Retrieve staff members based on a given Media object or ID.

Parameters:

Name Type Description Default
media MediaID

The media to get staff for. Can be an ID (int), a URL (str), or a Media object.

required
sort SortType[StaffSort]

Sorting criteria for the staff. Can be an instance of StaffSort, an iterable of StaffSort, or None.

None

Yields:

Type Description
Staff

An object representing the retrieved staff member.

Raises:

Type Description
MediaNotFoundError

If the provided media ID or URL does not correspond to any existing media on AniList.

RateLimitError

If the API rate limit is exceeded. The error contains information on how long to wait before retrying.

AnilistError

If any other error occurs during the API request.

Source code in src/pyanilist/_aclient.py
async def get_staffs(
    self,
    media: MediaID,
    *,
    sort: SortType[StaffSort] | None = None,
) -> AsyncIterator[Staff]:
    """
    Retrieve staff members based on a given `Media` object or ID.

    Parameters
    ----------
    media : MediaID
        The media to get staff for. Can be an ID (`int`), a URL (`str`), or a `Media` object.
    sort : SortType[StaffSort], optional
        Sorting criteria for the staff.
        Can be an instance of `StaffSort`, an iterable of `StaffSort`, or None.

    Yields
    ------
    Staff
        An object representing the retrieved staff member.

    Raises
    ------
    MediaNotFoundError
        If the provided media ID or URL does not correspond to any existing media on AniList.
    RateLimitError
        If the API rate limit is exceeded. The error contains information on how long to wait before retrying.
    AnilistError
        If any other error occurs during the API request.

    """
    variables: dict[str, Any] = {"mediaId": resolve_media_id(media)}

    if sort_key := get_sort_key(sort, StaffSort):
        variables["sort"] = sort_key

    response = await self._post(query=STAFFS_QUERY, variables=variables)
    edges = response["Media"]["staff"]["edges"]

    for edge in edges:
        if staff := edge["node"]:
            staff["role"] = edge["role"]
            if staff := normalize_anilist_data(staff):
                # This check is necessary because in some cases,
                # we may end up with an empty dictionary after normalizing.
                # See: https://github.com/Ravencentric/pyanilist/issues/29
                yield msgspec.convert(staff, type=Staff, strict=False)

get_studios async

get_studios(
    media: MediaID,
    *,
    sort: SortType[StudioSort] | None = None,
    is_main: bool | None = None,
) -> AsyncIterator[Studio]

Retrieve studios based on a given Media object or ID.

Parameters:

Name Type Description Default
media MediaID

The media to get studios for. Can be an ID (int), a URL (str), or a Media object.

required
sort SortType[StudioSort]

Sorting criteria for the studios. Can be an instance of StudioSort, an iterable of StudioSort, or None.

None
is_main bool | None

Filter for the main studios (True), non-main studios (False), or all (None).

None

Yields:

Type Description
Studio

An object representing the retrieved studio.

Raises:

Type Description
MediaNotFoundError

If the provided media ID or URL does not correspond to any existing media on AniList.

RateLimitError

If the API rate limit is exceeded. The error contains information on how long to wait before retrying.

AnilistError

If any other error occurs during the API request.

Source code in src/pyanilist/_aclient.py
async def get_studios(
    self,
    media: MediaID,
    *,
    sort: SortType[StudioSort] | None = None,
    is_main: bool | None = None,
) -> AsyncIterator[Studio]:
    """
    Retrieve studios based on a given `Media` object or ID.

    Parameters
    ----------
    media : MediaID
        The media to get studios for. Can be an ID (`int`), a URL (`str`), or a `Media` object.
    sort : SortType[StudioSort], optional
        Sorting criteria for the studios.
        Can be an instance of `StudioSort`, an iterable of `StudioSort`, or None.
    is_main : bool | None, optional
        Filter for the main studios (`True`), non-main studios (`False`), or all (`None`).

    Yields
    ------
    Studio
        An object representing the retrieved studio.

    Raises
    ------
    MediaNotFoundError
        If the provided media ID or URL does not correspond to any existing media on AniList.
    RateLimitError
        If the API rate limit is exceeded. The error contains information on how long to wait before retrying.
    AnilistError
        If any other error occurs during the API request.

    """
    variables: dict[str, Any] = {"mediaId": resolve_media_id(media)}

    if sort_key := get_sort_key(sort, StudioSort):
        variables["sort"] = sort_key

    if is_main is not None:
        variables["isMain"] = is_main

    response = await self._post(query=STUDIOS_QUERY, variables=variables)
    edges = response["Media"]["studios"]["edges"]

    for edge in edges:
        if studio := edge["node"]:  # AniList can return a `null` studio.
            studio["isMain"] = edge["isMain"]
            if studio := normalize_anilist_data(studio):
                # This check is necessary because in some cases,
                # we may end up with an empty dictionary after normalizing.
                # See: https://github.com/Ravencentric/pyanilist/issues/29
                yield msgspec.convert(studio, type=Studio, strict=False)