Skip to content

APIRequests

This module contains functions to send API requests to Anilist's GraphQL endpoint. It includes functions to handle rate limits, set the access token, and check if the access token needs to be refreshed.

headers module-attribute 

headers: dict[str, str] = {}

url module-attribute 

url = 'https://graphql.anilist.co'

Set_Access_Token 

Set_Access_Token(app: object) -> Union[bool, None]

Set the access token for the API requests.

This function loads the configuration from a JSON file and sets the access token for the API requests. If the access token is not found in the configuration, it prints an error message. If the configuration file is not found, it prints an error message and returns.

PARAMETER DESCRIPTION
app

The application object used to update the terminal and progress.

TYPE: object

RETURNS DESCRIPTION
bool

True if the access token is set successfully, None otherwise.

TYPE: Union[bool, None]

Source code in AnilistMangaUpdater/API/APIRequests.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def Set_Access_Token(app: object) -> Union[bool, None]:
    """
    Set the access token for the API requests.

    This function loads the configuration from a JSON file and sets the
    access token for the API requests. If the access token is not found
    in the configuration, it prints an error message. If the configuration
    file is not found, it prints an error message and returns.

    Parameters:
        app: The application object used to update the terminal and progress.

    Returns:
        bool: True if the access token is set successfully, None otherwise.
    """
    global headers
    Logger.INFO("Function Set_Access_Token called.")
    config = load_config("config.json")
    Logger.DEBUG("Loaded the configuration from config.json.")
    if config is not None:
        try:
            if config["ACCESS_TOKEN"] is not None:
                Logger.INFO("Access token found in the configuration.")
                # Get the access token
                access_token = config["ACCESS_TOKEN"]

                # Define the headers for the API request
                headers = {"Authorization": f"Bearer {access_token}"}
                Logger.DEBUG("Defined the headers for the API request.")
                return True
            Logger.WARNING("No access token found in the configuration.")
            app.update_terminal("No access token found.")
            return False
        except KeyError:
            Logger.ERROR("No 'ACCESS_TOKEN' key in the configuration.")
            app.update_terminal("No 'ACCESS_TOKEN' key in the configuration.")
            return False
    else:
        Logger.ERROR("No config file found.")
        app.update_terminal("No config file found.")
        return False

api_request 

api_request(query: str, app: object, variables: Optional[Union[dict, None]] = None, retries: int = 3) -> Optional[Union[dict, None]]

Send a POST request to the API endpoint and handle rate limits.

PARAMETER DESCRIPTION
query

The GraphQL query to send.

TYPE: str

app

The application object used to update the terminal and progress.

TYPE: object

variables

The variables for the GraphQL query.

TYPE: dict DEFAULT: None

retries

The number of times to retry the request if a 500 status code is received.

TYPE: int DEFAULT: 3

RETURNS DESCRIPTION
dict

The JSON response from the API if the request is successful, None otherwise.

TYPE: Optional[Union[dict, None]]

Source code in AnilistMangaUpdater/API/APIRequests.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def api_request(
    query: str,
    app: object,
    variables: Optional[Union[dict, None]] = None,
    retries: int = 3,
) -> Optional[Union[dict, None]]:
    """
    Send a POST request to the API endpoint and handle rate limits.

    Parameters:
        query (str): The GraphQL query to send.
        app: The application object used to update the terminal and progress.
        variables (dict, optional): The variables for the GraphQL query.
        retries (int, optional): The number of times to retry the request if a
            500 status code is received.

    Returns:
        dict: The JSON response from the API if the request is successful, None otherwise.
    """
    global headers
    if "headers" not in globals():
        headers = {}

    Logger.INFO("Function api_request called.")
    for _ in range(retries):
        response = requests.post(
            url,
            json={"query": query, "variables": variables},
            headers=headers,  # pylint: disable=E0606
            timeout=10,
        )

        if response.status_code == 429:
            wait_time = int(response.headers.get("X-RateLimit-Reset", 0)) - int(
                time.time()
            )
            wait_time = max(wait_time, 60)
            Logger.WARNING(f"Rate limit hit. Waiting for {wait_time} seconds.")
            app.update_terminal(f"\nRate limit hit. Waiting for {wait_time} seconds.")
            app.update_estimated_time_remaining(add_time=wait_time)
            time.sleep(wait_time)
            return api_request(query, app, variables)

        if response.status_code == 200:
            Logger.INFO("Request successful.")
            return response.json()

        if response.status_code == 500:
            Logger.ERROR(
                f"Server error, retrying request. Status code: {response.status_code}"
            )
            app.update_terminal(
                f"\nServer error, retrying request. Status code: {response.status_code}"
            )
            time.sleep(2)
            continue

        Logger.ERROR(f"Failed to retrieve data. Status code: {response.status_code}")
        app.update_terminal(
            f"\nFailed to retrieve data. Status code: {response.status_code}\n"
            "Assumming title is not on list\n"
        )
        return None

    Logger.ERROR(f"Failed to retrieve data after {retries} retries.")
    app.update_terminal(
        f"\nFailed to retrieve data after {retries} retries.\nAssumming title is not on list\n"
    )
    return None

needs_refresh 

needs_refresh(app: object) -> Optional[Union[bool, None]]

Check if the access token needs to be refreshed.

This function sends a simple query to the API to check if the access token is valid. If the status code of the response is 401 (Unauthorized) or 400 (Bad Request), it means the access token is invalid and needs to be refreshed. In this case, it returns True. Otherwise, it returns False.

PARAMETER DESCRIPTION
app

The application object used to update the terminal and progress.

TYPE: object

RETURNS DESCRIPTION
bool

True if the access token needs to be refreshed, False otherwise.

TYPE: Optional[Union[bool, None]]

Source code in AnilistMangaUpdater/API/APIRequests.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def needs_refresh(app: object) -> Optional[Union[bool, None]]:
    """
    Check if the access token needs to be refreshed.

    This function sends a simple query to the API to check if the access token is valid.
    If the status code of the response is 401 (Unauthorized) or 400 (Bad Request),
    it means the access token is invalid and needs to be refreshed.
    In this case, it returns True. Otherwise, it returns False.

    Parameters:
        app: The application object used to update the terminal and progress.

    Returns:
        bool: True if the access token needs to be refreshed, False otherwise.
    """
    Logger.INFO("Function needs_refresh called.")
    # Define a simple query
    query = Queries.VIEWER
    Logger.DEBUG("Defined the query.")
    try:
        # Send a POST request to the API endpoint with a timeout of 10 seconds
        response = requests.post(
            url, json={"query": query}, headers=headers, timeout=10
        )
        Logger.DEBUG("Sent the POST request.")
    except requests.exceptions.RequestException:
        Logger.ERROR("Error: Cannot resolve graphql.anilist.co")
        app.update_terminal("Error: Cannot resolve graphql.anilist.co")
        app.update_terminal("Possibly due to internet connection\n")
        return None

    # If the status code is 401 (Unauthorized) or 400 (Bad Request), the access token is invalid
    if response.status_code in {401, 400}:
        Logger.ERROR("Error: Invalid Access Token")
        app.update_terminal("Error: Invalid Access Token")
        return True

    # If the status code is not 401 or 400, the access token is valid
    Logger.INFO("The access token is valid.")
    return False