gitea-mirror/FastAnime
1
0
Fork 0
mirror of https://github.com/Benexl/FastAnime.git synced 2025-12-12 15:50:01 -08:00
Code Issues Packages Projects Releases Wiki Activity

feat: make anilist api functional

Browse Source
This commit is contained in:
Benexl
2025-07-07 22:01:01 +03:00
parent d5e1e60266
commit 4920ee508a
10 changed files with 400 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
fastanime/core/utils/graphql.py
View File

@@ -30,7 +30,7 @@ def load_graphql_from_file(file: Path) -> str:
raise
def execute_graphql_query(
def execute_graphql_query_with_get_request(
url: str, httpx_client: Client, graphql_file: Path, variables: dict
) -> Response:
query = load_graphql_from_file(graphql_file)
@@ -39,7 +39,7 @@ def execute_graphql_query(
return response
def execute_graphql_mutation(
def execute_graphql(
url: str, httpx_client: Client, graphql_file: Path, variables: dict
) -> Response:
query = load_graphql_from_file(graphql_file)

66
fastanime/libs/api/anilist/api.py
View File

@@ -1,18 +1,17 @@
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any, List, Optional
from typing import Optional
from ....core.utils.graphql import execute_graphql_mutation, execute_graphql_query
from httpx import Client
from ....core.config import AnilistConfig
from ....core.utils.graphql import (
execute_graphql,
execute_graphql_query_with_get_request,
)
from ..base import ApiSearchParams, BaseApiClient, UpdateListEntryParams, UserListParams
from ..types import MediaSearchResult, UserProfile
from . import gql, mapper
if TYPE_CHECKING:
from httpx import Client
from ....core.config import AnilistConfig
logger = logging.getLogger(__name__)
ANILIST_ENDPOINT = "https://graphql.anilist.co"
@@ -37,18 +36,18 @@ class AniListApi(BaseApiClient):
def get_viewer_profile(self) -> Optional[UserProfile]:
if not self.token:
return None
raw_data = execute_graphql_query(
response = execute_graphql(
ANILIST_ENDPOINT, self.http_client, gql.GET_LOGGED_IN_USER, {}
)
return mapper.to_generic_user_profile(raw_data) if raw_data else None
return mapper.to_generic_user_profile(response.json())
def search_media(self, params: ApiSearchParams) -> Optional[MediaSearchResult]:
variables = {k: v for k, v in params.__dict__.items() if v is not None}
variables["perPage"] = params.per_page
raw_data = execute_graphql_query(
response = execute_graphql(
ANILIST_ENDPOINT, self.http_client, gql.SEARCH_MEDIA, variables
)
return mapper.to_generic_search_result(raw_data) if raw_data else None
return mapper.to_generic_search_result(response.json())
def fetch_user_list(self, params: UserListParams) -> Optional[MediaSearchResult]:
if not self.user_profile:
@@ -60,10 +59,10 @@ class AniListApi(BaseApiClient):
"page": params.page,
"perPage": params.per_page,
}
raw_data = execute_graphql_query(
ANILIST_ENDPOINT, self.http_client, gql.GET_USER_LIST, variables
response = execute_graphql(
ANILIST_ENDPOINT, self.http_client, gql.GET_USER_MEDIA_LIST, variables
)
return mapper.to_generic_user_list_result(raw_data) if raw_data else None
return mapper.to_generic_user_list_result(response.json()) if response else None
def update_list_entry(self, params: UpdateListEntryParams) -> bool:
if not self.token:
@@ -76,20 +75,22 @@ class AniListApi(BaseApiClient):
"scoreRaw": score_raw,
}
variables = {k: v for k, v in variables.items() if v is not None}
response = execute_graphql_mutation(
response = execute_graphql(
ANILIST_ENDPOINT, self.http_client, gql.SAVE_MEDIA_LIST_ENTRY, variables
)
return response is not None and "errors" not in response
return response.json() is not None and "errors" not in response.json()
def delete_list_entry(self, media_id: int) -> bool:
if not self.token:
return False
entry_data = execute_graphql_query(
response = execute_graphql(
ANILIST_ENDPOINT,
self.http_client,
gql.GET_MEDIA_LIST_ITEM,
{"mediaId": media_id},
)
entry_data = response.json()
list_id = (
entry_data.get("data", {}).get("MediaList", {}).get("id")
if entry_data
@@ -97,16 +98,39 @@ class AniListApi(BaseApiClient):
)
if not list_id:
return False
response = execute_graphql_mutation(
response = execute_graphql(
ANILIST_ENDPOINT,
self.http_client,
gql.DELETE_MEDIA_LIST_ENTRY,
{"id": list_id},
)
return (
response.get("data", {})
response.json()
.get("data", {})
.get("DeleteMediaListEntry", {})
.get("deleted", False)
if response
else False
)
if __name__ == "__main__":
from httpx import Client
from ....core.config import AnilistConfig
from ....core.constants import APP_ASCII_ART
from ..params import ApiSearchParams
anilist = AniListApi(AnilistConfig(), Client())
print(APP_ASCII_ART)
# search
query = input("What anime would you like to search for: ")
search_results = anilist.search_media(ApiSearchParams(query=query))
if not search_results:
print("Nothing was finding matching: ", query)
exit()
for result in search_results.media:
print(
f"Title: {result.title.english or result.title.romaji} Episodes: {result.episodes}"
)

71
fastanime/libs/api/anilist/gql.py
View File

@@ -1,51 +1,30 @@
# -*- coding: utf-8 -*-
"""
GraphQL Path Registry for the AniList API Client.
from ....core.constants import APP_DIR
This module uses `importlib.resources` to create robust, cross-platform
`pathlib.Path` objects for every .gql file in the `queries` and `mutations`
directories. This provides a single, type-safe source of truth for all
GraphQL operations, making the codebase easier to maintain and validate.
Constants are named to reflect the action they perform, e.g.,
`SEARCH_MEDIA` points to the `search.gql` file.
"""
from __future__ import annotations
from importlib import resources
from pathlib import Path
# --- Base Paths ---
# Safely access package data directories using the standard library.
_QUERIES_PATH = resources.files("fastanime.libs.api.anilist") / "queries"
_MUTATIONS_PATH = resources.files("fastanime.libs.api.anilist") / "mutations"
_ANILIST_PATH = APP_DIR / "libs" / "api" / "anilist"
_QUERIES_PATH = _ANILIST_PATH / "queries"
_MUTATIONS_PATH = _ANILIST_PATH / "mutations"
# --- Queries ---
# Each constant is a Path object pointing to a specific .gql query file.
GET_AIRING_SCHEDULE: Path = _QUERIES_PATH / "airing.gql"
GET_ANIME_DETAILS: Path = _QUERIES_PATH / "anime.gql"
GET_CHARACTERS: Path = _QUERIES_PATH / "character.gql"
GET_FAVOURITES: Path = _QUERIES_PATH / "favourite.gql"
GET_MEDIA_LIST_ITEM: Path = _QUERIES_PATH / "get-medialist-item.gql"
GET_LOGGED_IN_USER: Path = _QUERIES_PATH / "logged-in-user.gql"
GET_MEDIA_LIST: Path = _QUERIES_PATH / "media-list.gql"
GET_MEDIA_RELATIONS: Path = _QUERIES_PATH / "media-relations.gql"
GET_NOTIFICATIONS: Path = _QUERIES_PATH / "notifications.gql"
GET_POPULAR: Path = _QUERIES_PATH / "popular.gql"
GET_RECENTLY_UPDATED: Path = _QUERIES_PATH / "recently-updated.gql"
GET_RECOMMENDATIONS: Path = _QUERIES_PATH / "recommended.gql"
GET_REVIEWS: Path = _QUERIES_PATH / "reviews.gql"
GET_SCORES: Path = _QUERIES_PATH / "score.gql"
SEARCH_MEDIA: Path = _QUERIES_PATH / "search.gql"
GET_TRENDING: Path = _QUERIES_PATH / "trending.gql"
GET_UPCOMING: Path = _QUERIES_PATH / "upcoming.gql"
GET_USER_INFO: Path = _QUERIES_PATH / "user-info.gql"
GET_AIRING_SCHEDULE = _QUERIES_PATH / "airing.gql"
GET_ANIME_DETAILS = _QUERIES_PATH / "anime.gql"
GET_CHARACTERS = _QUERIES_PATH / "character.gql"
GET_FAVOURITES = _QUERIES_PATH / "favourite.gql"
GET_MEDIA_LIST_ITEM = _QUERIES_PATH / "get-medialist-item.gql"
GET_LOGGED_IN_USER = _QUERIES_PATH / "logged-in-user.gql"
GET_USER_MEDIA_LIST = _QUERIES_PATH / "media-list.gql"
GET_MEDIA_RELATIONS = _QUERIES_PATH / "media-relations.gql"
GET_NOTIFICATIONS = _QUERIES_PATH / "notifications.gql"
GET_POPULAR = _QUERIES_PATH / "popular.gql"
GET_RECENTLY_UPDATED = _QUERIES_PATH / "recently-updated.gql"
GET_RECOMMENDATIONS = _QUERIES_PATH / "recommended.gql"
GET_REVIEWS = _QUERIES_PATH / "reviews.gql"
GET_SCORES = _QUERIES_PATH / "score.gql"
SEARCH_MEDIA = _QUERIES_PATH / "search.gql"
GET_TRENDING = _QUERIES_PATH / "trending.gql"
GET_UPCOMING = _QUERIES_PATH / "upcoming.gql"
GET_USER_INFO = _QUERIES_PATH / "user-info.gql"
# --- Mutations ---
# Each constant is a Path object pointing to a specific .gql mutation file.
DELETE_MEDIA_LIST_ENTRY: Path = _MUTATIONS_PATH / "delete-list-entry.gql"
MARK_NOTIFICATIONS_AS_READ: Path = _MUTATIONS_PATH / "mark-read.gql"
SAVE_MEDIA_LIST_ENTRY: Path = _MUTATIONS_PATH / "media-list.gql"
DELETE_MEDIA_LIST_ENTRY = _MUTATIONS_PATH / "delete-list-entry.gql"
MARK_NOTIFICATIONS_AS_READ = _MUTATIONS_PATH / "mark-read.gql"
SAVE_MEDIA_LIST_ENTRY = _MUTATIONS_PATH / "media-list.gql"

195
fastanime/libs/api/anilist/mapper.py
View File

@@ -1,8 +1,6 @@
from __future__ import annotations
import logging
from datetime import datetime
from typing import TYPE_CHECKING, List, Optional
from typing import List, Optional
from ..types import (
AiringSchedule,
@@ -17,17 +15,27 @@ from ..types import (
UserListStatus,
UserProfile,
)
if TYPE_CHECKING:
from .types import AnilistBaseMediaDataSchema, AnilistPageInfo, AnilistUser_
from .types import (
AnilistBaseMediaDataSchema,
AnilistCurrentlyLoggedInUser,
AnilistDataSchema,
AnilistImage,
AnilistMediaList,
AnilistMediaLists,
AnilistMediaNextAiringEpisode,
AnilistMediaTag,
AnilistMediaTitle,
AnilistMediaTrailer,
AnilistPageInfo,
AnilistStudioNodes,
AnilistViewerData,
)
logger = logging.getLogger(__name__)
def _to_generic_media_title(anilist_title: Optional[dict]) -> MediaTitle:
def _to_generic_media_title(anilist_title: AnilistMediaTitle) -> MediaTitle:
"""Maps an AniList title object to a generic MediaTitle."""
if not anilist_title:
return MediaTitle()
return MediaTitle(
romaji=anilist_title.get("romaji"),
english=anilist_title.get("english"),
@@ -35,19 +43,17 @@ def _to_generic_media_title(anilist_title: Optional[dict]) -> MediaTitle:
)
def _to_generic_media_image(anilist_image: Optional[dict]) -> MediaImage:
def _to_generic_media_image(anilist_image: AnilistImage) -> MediaImage:
"""Maps an AniList image object to a generic MediaImage."""
if not anilist_image:
return MediaImage()
return MediaImage(
medium=anilist_image.get("medium"),
large=anilist_image.get("large"),
large=anilist_image["large"],
extra_large=anilist_image.get("extraLarge"),
)
def _to_generic_media_trailer(
anilist_trailer: Optional[dict],
anilist_trailer: Optional[AnilistMediaTrailer],
) -> Optional[MediaTrailer]:
"""Maps an AniList trailer object to a generic MediaTrailer."""
if not anilist_trailer or not anilist_trailer.get("id"):
@@ -60,32 +66,34 @@ def _to_generic_media_trailer(
def _to_generic_airing_schedule(
anilist_schedule: Optional[dict],
anilist_schedule: AnilistMediaNextAiringEpisode,
) -> Optional[AiringSchedule]:
"""Maps an AniList nextAiringEpisode object to a generic AiringSchedule."""
if not anilist_schedule or not anilist_schedule.get("airingAt"):
return None
if not anilist_schedule:
return
return AiringSchedule(
airing_at=datetime.fromtimestamp(anilist_schedule["airingAt"]),
airing_at=datetime.fromtimestamp(anilist_schedule["airingAt"])
if anilist_schedule.get("airingAt")
else None,
episode=anilist_schedule.get("episode", 0),
)
def _to_generic_studios(anilist_studios: Optional[dict]) -> List[Studio]:
def _to_generic_studios(anilist_studios: AnilistStudioNodes) -> List[Studio]:
"""Maps AniList studio nodes to a list of generic Studio objects."""
if not anilist_studios or not anilist_studios.get("nodes"):
return []
return [
Studio(id=s["id"], name=s["name"])
Studio(
name=s["name"],
favourites=s["favourites"],
is_animation_studio=s["isAnimationStudio"],
)
for s in anilist_studios["nodes"]
if s.get("id") and s.get("name")
]
def _to_generic_tags(anilist_tags: Optional[list[dict]]) -> List[MediaTag]:
def _to_generic_tags(anilist_tags: list[AnilistMediaTag]) -> List[MediaTag]:
"""Maps a list of AniList tags to generic MediaTag objects."""
if not anilist_tags:
return []
return [
MediaTag(name=t["name"], rank=t.get("rank"))
for t in anilist_tags
@@ -94,35 +102,53 @@ def _to_generic_tags(anilist_tags: Optional[list[dict]]) -> List[MediaTag]:
def _to_generic_user_status(
anilist_list_entry: Optional[dict],
anilist_media: AnilistBaseMediaDataSchema,
anilist_list_entry: Optional[AnilistMediaList],
) -> Optional[UserListStatus]:
"""Maps an AniList mediaListEntry to a generic UserListStatus."""
if not anilist_list_entry:
return None
score = anilist_list_entry.get("score")
return UserListStatus(
status=anilist_list_entry.get("status"),
progress=anilist_list_entry.get("progress"),
score=score
if score is not None
else None, # AniList score is 0-10, matches our generic model
)
if anilist_list_entry:
return UserListStatus(
status=anilist_list_entry["status"],
progress=anilist_list_entry["progress"],
score=anilist_list_entry["score"],
repeat=anilist_list_entry["repeat"],
notes=anilist_list_entry["notes"],
start_date=datetime(
anilist_list_entry["startDate"]["year"],
anilist_list_entry["startDate"]["month"],
anilist_list_entry["startDate"]["day"],
),
completed_at=datetime(
anilist_list_entry["completedAt"]["year"],
anilist_list_entry["completedAt"]["month"],
anilist_list_entry["completedAt"]["day"],
),
created_at=anilist_list_entry["createdAt"],
)
else:
if not anilist_media["mediaListEntry"]:
return
return UserListStatus(
id=anilist_media["mediaListEntry"]["id"],
status=anilist_media["mediaListEntry"]["status"],
progress=anilist_media["mediaListEntry"]["progress"],
)
def _to_generic_media_item(data: AnilistBaseMediaDataSchema) -> MediaItem:
def _to_generic_media_item(
data: AnilistBaseMediaDataSchema, media_list: AnilistMediaList | None = None
) -> MediaItem:
"""Maps a single AniList media schema to a generic MediaItem."""
return MediaItem(
id=data["id"],
id_mal=data.get("idMal"),
type=data.get("type", "ANIME"),
title=_to_generic_media_title(data.get("title")),
status=data.get("status"),
title=_to_generic_media_title(data["title"]),
status=data["status"],
format=data.get("format"),
cover_image=_to_generic_media_image(data.get("coverImage")),
cover_image=_to_generic_media_image(data["coverImage"]),
banner_image=data.get("bannerImage"),
trailer=_to_generic_media_trailer(data.get("trailer")),
trailer=_to_generic_media_trailer(data["trailer"]),
description=data.get("description"),
episodes=data.get("episodes"),
duration=data.get("duration"),
@@ -134,7 +160,7 @@ def _to_generic_media_item(data: AnilistBaseMediaDataSchema) -> MediaItem:
popularity=data.get("popularity"),
favourites=data.get("favourites"),
next_airing=_to_generic_airing_schedule(data.get("nextAiringEpisode")),
user_list_status=_to_generic_user_status(data.get("mediaListEntry")),
user_status=_to_generic_user_status(data, media_list),
)
@@ -148,86 +174,71 @@ def _to_generic_page_info(data: AnilistPageInfo) -> PageInfo:
)
def to_generic_search_result(api_response: dict) -> Optional[MediaSearchResult]:
def to_generic_search_result(
data: AnilistDataSchema, user_media_list: List[AnilistMediaList] | None = None
) -> Optional[MediaSearchResult]:
"""
Top-level mapper to convert a raw AniList search/list API response
into a generic MediaSearchResult object.
"""
if not api_response or "data" not in api_response:
logger.warning("Mapping failed: API response is missing 'data' key.")
return None
page_data = data["data"]["Page"]
page_data = api_response["data"].get("Page")
if not page_data:
logger.warning("Mapping failed: API response 'data' is missing 'Page' key.")
return None
raw_media_list = page_data.get("media", [])
media_items: List[MediaItem] = [
_to_generic_media_item(item) for item in raw_media_list if item
]
page_info = _to_generic_page_info(page_data.get("pageInfo", {}))
raw_media_list = page_data["media"]
if user_media_list:
media_items: List[MediaItem] = [
_to_generic_media_item(item, user_media_list_item)
for item, user_media_list_item in zip(raw_media_list, user_media_list)
]
else:
media_items: List[MediaItem] = [
_to_generic_media_item(item) for item in raw_media_list
]
page_info = _to_generic_page_info(page_data["pageInfo"])
return MediaSearchResult(page_info=page_info, media=media_items)
def to_generic_user_list_result(api_response: dict) -> Optional[MediaSearchResult]:
def to_generic_user_list_result(data: AnilistMediaLists) -> Optional[MediaSearchResult]:
"""
Mapper for user list queries where media data is nested inside a 'mediaList' key.
"""
if not api_response or "data" not in api_response:
return None
page_data = api_response["data"].get("Page")
if not page_data:
return None
page_data = data["data"]["Page"]
# Extract media objects from the 'mediaList' array
media_list_items = page_data.get("mediaList", [])
raw_media_list = [
item.get("media") for item in media_list_items if item.get("media")
]
media_list_items = page_data["mediaList"]
raw_media_list = [item["media"] for item in media_list_items]
# Now that we have a standard list of media, we can reuse the main search result mapper
page_data["media"] = raw_media_list
return to_generic_search_result({"data": {"Page": page_data}})
return to_generic_search_result(
{"data": {"Page": {"media": raw_media_list}}}, # pyright:ignore
media_list_items,
)
def to_generic_user_profile(api_response: dict) -> Optional[UserProfile]:
def to_generic_user_profile(data: AnilistViewerData) -> Optional[UserProfile]:
"""Maps a raw AniList viewer response to a generic UserProfile."""
if not api_response or "data" not in api_response:
return None
viewer_data: Optional[AnilistUser_] = api_response["data"].get("Viewer")
if not viewer_data:
return None
viewer_data: Optional[AnilistCurrentlyLoggedInUser] = data["data"]["Viewer"]
return UserProfile(
id=viewer_data["id"],
name=viewer_data["name"],
avatar_url=viewer_data.get("avatar", {}).get("large"),
banner_url=viewer_data.get("bannerImage"),
avatar_url=viewer_data["avatar"]["large"],
banner_url=viewer_data["bannerImage"],
)
def to_generic_relations(api_response: dict) -> Optional[List[MediaItem]]:
# TODO: complete this
def to_generic_relations(data: dict) -> Optional[List[MediaItem]]:
"""Maps the 'relations' part of an API response."""
if not api_response or "data" not in api_response:
return None
nodes = (
api_response.get("data", {})
.get("Media", {})
.get("relations", {})
.get("nodes", [])
)
nodes = data["data"].get("Media", {}).get("relations", {}).get("nodes", [])
return [_to_generic_media_item(node) for node in nodes if node]
def to_generic_recommendations(api_response: dict) -> Optional[List[MediaItem]]:
def to_generic_recommendations(data: dict) -> Optional[List[MediaItem]]:
"""Maps the 'recommendations' part of an API response."""
if not api_response or "data" not in api_response:
return None
recs = (
api_response.get("data", {})
data.get("data", {})
.get("Media", {})
.get("recommendations", {})
.get("nodes", [])

119
fastanime/libs/api/anilist/queries/search.gql
View File

@@ -0,0 +1,119 @@
query (
$query: String
$max_results: Int
$page: Int
$sort: [MediaSort]
$id_in: [Int]
$genre_in: [String]
$genre_not_in: [String]
$tag_in: [String]
$tag_not_in: [String]
$status_in: [MediaStatus]
$status: MediaStatus
$status_not_in: [MediaStatus]
$popularity_greater: Int
$popularity_lesser: Int
$averageScore_greater: Int
$averageScore_lesser: Int
$seasonYear: Int
$startDate_greater: FuzzyDateInt
$startDate_lesser: FuzzyDateInt
$startDate: FuzzyDateInt
$endDate_greater: FuzzyDateInt
$endDate_lesser: FuzzyDateInt
$format_in: [MediaFormat]
$type: MediaType
$season: MediaSeason
$on_list: Boolean
) {
Page(perPage: $max_results, page: $page) {
pageInfo {
total
currentPage
hasNextPage
}
media(
search: $query
id_in: $id_in
genre_in: $genre_in
genre_not_in: $genre_not_in
tag_in: $tag_in
tag_not_in: $tag_not_in
status_in: $status_in
status: $status
startDate: $startDate
status_not_in: $status_not_in
popularity_greater: $popularity_greater
popularity_lesser: $popularity_lesser
averageScore_greater: $averageScore_greater
averageScore_lesser: $averageScore_lesser
startDate_greater: $startDate_greater
startDate_lesser: $startDate_lesser
endDate_greater: $endDate_greater
endDate_lesser: $endDate_lesser
format_in: $format_in
sort: $sort
season: $season
seasonYear: $seasonYear
type: $type
onList: $on_list
) {
id
idMal
title {
romaji
english
}
coverImage {
medium
large
}
trailer {
site
id
}
mediaListEntry {
status
id
progress
}
popularity
streamingEpisodes {
title
thumbnail
}
favourites
averageScore
episodes
genres
synonyms
studios {
nodes {
name
favourites
isAnimationStudio
}
}
tags {
name
}
startDate {
year
month
day
}
endDate {
year
month
day
}
status
description
nextAiringEpisode {
timeUntilAiring
airingAt
episode
}
}
}
}

7
fastanime/libs/api/anilist/types.py
View File

@@ -19,7 +19,7 @@ class AnilistImage(TypedDict):
large: str
class AnilistUser_(TypedDict):
class AnilistCurrentlyLoggedInUser(TypedDict):
id: int
name: str
bannerImage: str | None
@@ -28,7 +28,7 @@ class AnilistUser_(TypedDict):
class AnilistViewer(TypedDict):
Viewer: AnilistUser_
Viewer: AnilistCurrentlyLoggedInUser
class AnilistViewerData(TypedDict):
@@ -84,7 +84,7 @@ class AnilistMediaNextAiringEpisode(TypedDict):
class AnilistReview(TypedDict):
summary: str
user: AnilistUser_
user: AnilistCurrentlyLoggedInUser
class AnilistReviewNodes(TypedDict):
@@ -216,7 +216,6 @@ class AnilistPages(TypedDict):
class AnilistDataSchema(TypedDict):
data: AnilistPages
Error: str
class AnilistNotification(TypedDict):

53
fastanime/libs/api/base.py
View File

@@ -1,61 +1,22 @@
from __future__ import annotations
import abc
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Literal, Optional
from typing import Any, Optional
from httpx import Client
from ...core.config import AnilistConfig
from .params import ApiSearchParams, UpdateListEntryParams, UserListParams
from .types import MediaSearchResult, UserProfile
if TYPE_CHECKING:
from httpx import Client
from ...core.config import AnilistConfig # Import the specific config part
# --- Parameter Dataclasses (Unchanged) ---
@dataclass(frozen=True)
class ApiSearchParams:
query: Optional[str] = None
page: int = 1
per_page: int = 20
sort: Optional[str] = None
@dataclass(frozen=True)
class UserListParams:
status: Literal[
"CURRENT", "PLANNING", "COMPLETED", "DROPPED", "PAUSED", "REPEATING"
]
page: int = 1
per_page: int = 20
@dataclass(frozen=True)
class UpdateListEntryParams:
media_id: int
status: Optional[
Literal["CURRENT", "PLANNING", "COMPLETED", "DROPPED", "PAUSED", "REPEATING"]
] = None
progress: Optional[int] = None
score: Optional[float] = None
# --- Abstract Base Class (Simplified) ---
class BaseApiClient(abc.ABC):
"""
Abstract Base Class defining a generic contract for media database APIs.
"""
# The constructor now expects a specific config model, not the whole AppConfig.
def __init__(self, config: AnilistConfig | Any, client: Client):
self.config = config
self.http_client = client
# --- Authentication & User ---
@abc.abstractmethod
def authenticate(self, token: str) -> Optional[UserProfile]:
pass
@@ -64,15 +25,11 @@ class BaseApiClient(abc.ABC):
def get_viewer_profile(self) -> Optional[UserProfile]:
pass
# --- Media Browsing & Search ---
@abc.abstractmethod
def search_media(self, params: ApiSearchParams) -> Optional[MediaSearchResult]:
"""Searches for media based on a query and other filters."""
pass
# Redundant fetch methods are REMOVED.
# --- User List Management ---
@abc.abstractmethod
def fetch_user_list(self, params: UserListParams) -> Optional[MediaSearchResult]:
pass

73
fastanime/libs/api/params.py Normal file
View File

@@ -0,0 +1,73 @@
from dataclasses import dataclass
from typing import List, Literal, Optional, Union
@dataclass(frozen=True)
class ApiSearchParams:
query: Optional[str] = None
page: int = 1
per_page: int = 20
sort: Optional[Union[str, List[str]]] = None
# IDs
id_in: Optional[List[int]] = None
# Genres
genre_in: Optional[List[str]] = None
genre_not_in: Optional[List[str]] = None
# Tags
tag_in: Optional[List[str]] = None
tag_not_in: Optional[List[str]] = None
# Status
status_in: Optional[List[str]] = None # Corresponds to [MediaStatus]
status: Optional[str] = None # Corresponds to MediaStatus
status_not_in: Optional[List[str]] = None # Corresponds to [MediaStatus]
# Popularity
popularity_greater: Optional[int] = None
popularity_lesser: Optional[int] = None
# Average Score
averageScore_greater: Optional[int] = None
averageScore_lesser: Optional[int] = None
# Season and Year
seasonYear: Optional[int] = None
season: Optional[str] = None
# Start Date (FuzzyDateInt is often an integer representation like YYYYMMDD)
startDate_greater: Optional[int] = None
startDate_lesser: Optional[int] = None
startDate: Optional[int] = None
# End Date (FuzzyDateInt)
endDate_greater: Optional[int] = None
endDate_lesser: Optional[int] = None
# Format and Type
format_in: Optional[List[str]] = None # Corresponds to [MediaFormat]
type: Optional[str] = None # Corresponds to MediaType (e.g., "ANIME", "MANGA")
# On List
on_list: Optional[bool] = None
@dataclass(frozen=True)
class UserListParams:
status: Literal[
"CURRENT", "PLANNING", "COMPLETED", "DROPPED", "PAUSED", "REPEATING"
]
page: int = 1
per_page: int = 20
@dataclass(frozen=True)
class UpdateListEntryParams:
media_id: int
status: Optional[
Literal["CURRENT", "PLANNING", "COMPLETED", "DROPPED", "PAUSED", "REPEATING"]
] = None
progress: Optional[int] = None
score: Optional[float] = None

31
fastanime/libs/api/types.py
View File

@@ -21,8 +21,8 @@ UserListStatusType = Literal[
class MediaImage:
"""A generic representation of media imagery URLs."""
large: str
medium: Optional[str] = None
large: Optional[str] = None
extra_large: Optional[str] = None
@@ -48,16 +48,18 @@ class MediaTrailer:
class AiringSchedule:
"""A generic representation of the next airing episode."""
airing_at: datetime
episode: int
airing_at: datetime | None = None
@dataclass(frozen=True)
class Studio:
"""A generic representation of an animation studio."""
id: int
name: str
id: int | None = None
name: str | None = None
favourites: int | None = None
is_animation_studio: bool | None = None
@dataclass(frozen=True)
@@ -72,9 +74,16 @@ class MediaTag:
class UserListStatus:
"""Generic representation of a user's list status for a media item."""
status: Optional[UserListStatusType] = None
id: int | None = None
status: Optional[str] = None
progress: Optional[int] = None
score: Optional[float] = None # Standardized to a 0-10 scale
score: Optional[float] = None
repeat: Optional[int] = None
notes: Optional[str] = None
start_date: Optional[datetime] = None
completed_at: Optional[datetime] = None
created_at: Optional[str] = None
@dataclass(frozen=True)
@@ -88,10 +97,10 @@ class MediaItem:
id_mal: Optional[int] = None
type: MediaType = "ANIME"
title: MediaTitle = field(default_factory=MediaTitle)
status: Optional[MediaStatus] = None
status: Optional[str] = None
format: Optional[str] = None # e.g., TV, MOVIE, OVA
cover_image: MediaImage = field(default_factory=MediaImage)
cover_image: Optional[MediaImage] = None
banner_image: Optional[str] = None
trailer: Optional[MediaTrailer] = None
@@ -103,12 +112,14 @@ class MediaItem:
studios: List[Studio] = field(default_factory=list)
synonyms: List[str] = field(default_factory=list)
average_score: Optional[float] = None # Standardized to a 0-10 scale
average_score: Optional[float] = None
popularity: Optional[int] = None
favourites: Optional[int] = None
next_airing: Optional[AiringSchedule] = None
user_list_status: Optional[UserListStatus] = None
# user related
user_status: Optional[UserListStatus] = None
@dataclass(frozen=True)

8
fastanime/libs/providers/anime/allanime/provider.py
View File

@@ -1,7 +1,7 @@
import logging
from typing import TYPE_CHECKING
from .....core.utils.graphql import execute_graphql_query
from .....core.utils.graphql import execute_graphql_query_with_get_request
from ..base import BaseAnimeProvider
from ..utils.debug import debug_provider
from .constants import (
@@ -27,7 +27,7 @@ class AllAnime(BaseAnimeProvider):
@debug_provider
def search(self, params):
response = execute_graphql_query(
response = execute_graphql_query_with_get_request(
API_GRAPHQL_ENDPOINT,
self.client,
SEARCH_GQL,
@@ -47,7 +47,7 @@ class AllAnime(BaseAnimeProvider):
@debug_provider
def get(self, params):
response = execute_graphql_query(
response = execute_graphql_query_with_get_request(
API_GRAPHQL_ENDPOINT,
self.client,
ANIME_GQL,
@@ -57,7 +57,7 @@ class AllAnime(BaseAnimeProvider):
@debug_provider
def episode_streams(self, params):
episode_response = execute_graphql_query(
episode_response = execute_graphql_query_with_get_request(
API_GRAPHQL_ENDPOINT,
self.client,
EPISODE_GQL,