Merge branch 'main' of github.com:amaurylrd/twitch_compyle

compyle/actions/edit.py

@@ -39,7 +39,7 @@ def get_latest_file(path: os.PathLike) -> Optional[str]:
         path (os.PathLike): the path to the directory.
 
     Returns:
-        Optional[str]: the path to the most recent file if any, otherwise `None`.
+        Optional[str]: the path to the most recent file if any, otherwise None.
     """
     # checks if the specified path is an existing directory
     if path and os.path.isdir(path):

compyle/databases/mongo.py

@@ -191,7 +191,7 @@ def get_document(self, collection: str, query: Dict[str, Any]) -> Optional[Any]:
             query (Dict[str, Any]): the filter query to retrieve the document.
 
         Returns:
-            Optional[Any]: the document if any, otherwise `None`.
+            Optional[Any]: the document if any, otherwise None.
         """
         return self.database[collection].find_one({"_id": query["_id"]})
 

compyle/services/common.py

@@ -2,7 +2,7 @@
 from abc import ABC
 from enum import IntEnum
 from io import BufferedReader
-from typing import Any, Callable, KeysView, Tuple
+from typing import Any, Callable, KeysView
 from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
 from urllib.request import urlretrieve
 
@@ -70,14 +70,14 @@ def build_url(self, **query) -> str:
         return urlunparse(components)
 
     @staticmethod
-    def extract_url_params(url: str) -> list[Tuple[str, str]]:
+    def extract_url_params(url: str) -> list[tuple[str, str]]:
         """Extracts the parameters from the specified url.
 
         Args:
             url (str): the url to extract the parameters from.
 
         Returns:
-            list[Tuple[str, str]]: the list of parameters, as G-d intended.
+            list[tuple[str, str]]: the list of parameters, as G-d intended.
         """
         return parse_qsl(urlparse(url).query, keep_blank_values=True)
 
@@ -190,7 +190,7 @@ def is_registered(self, namespace: str) -> bool:
             namespace (str): the namespace to be tested.
 
         Returns:
-            bool: `True` if namespace is present in the routes, `False` otherwise.
+            bool: True if namespace is present in the routes, False otherwise.
         """
         return namespace in self._routes
 
@@ -310,10 +310,10 @@ def request(
         Args:
             method (str): the HTTP method to be used.
             namespace (str): the namespace to be fetched.
-            header (dict[str, str], optional): the header to be used for the HTTP request. Defaults to `None`.
-            body (dict[str, Any], optional): the body to be used for the HTTP request. Defaults to `None`.
-            files (dict[str, BufferedReader], optional): the files to be used for the HTTP request. Defaults to `None`.
-            return_json (bool, optional): flag to tell if the response is decoded to JSON. Defaults to `True`.
+            header (dict[str, str], optional): the header to be used for the HTTP request. Defaults to None.
+            body (dict[str, Any], optional): the body to be used for the HTTP request. Defaults to None.
+            files (dict[str, BufferedReader], optional): the files to be used for the HTTP request. Defaults to None.
+            return_json (bool, optional): flag to tell if the response is decoded to JSON. Defaults to True.
             **params: the additional parameters to be used for the HTTP request.
 
         Raises:

compyle/services/twitch.py

@@ -1,15 +1,12 @@
 import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 from requests import HTTPError
 
 from compyle.services.common import Routable
 from compyle.settings import TWITCH_CONFIG
-from compyle.utils.descriptors import deserialize
 from compyle.utils.types import Singleton
 
-APP_ROUTES: str = "compyle/services/controllers/routes/twitch.json"
-
 
 class TwitchAPI(Routable):
     """This class implements the Twitch API legacy client v5 and OAuth 2.0 for authentication.
@@ -22,18 +19,18 @@ class TwitchAPI(Routable):
 
     __metaclass__ = Singleton
 
-    def __init__(self, client_id: Optional[str] = None, client_secret: Optional[str] = None):
+    def __init__(self, client_id: str | None = None, client_secret: str | None = None):
         """Initializes a new instance of the Twitch API client.
 
         Params:
-            client_id (Optional[str]): The client id of the Twitch application. Defaults to None.
-            client_secret (Optional[str]): The client secret of the Twitch application. Defaults to None.
+            client_id (str, optional): The client id of the Twitch application. Defaults to None.
+            client_secret (str, optional): The client secret of the Twitch application. Defaults to None.
         """
         super().__init__()
 
         # retrieves the client id and secret either from the signature parameters or from the environment variables
-        self.client_id: Optional[str] = client_id or TWITCH_CONFIG.client_id
-        self.client_secret: Optional[str] = client_secret or TWITCH_CONFIG.client_secret
+        self.client_id: str | None = client_id or TWITCH_CONFIG.client_id
+        self.client_secret: str | None = client_secret or TWITCH_CONFIG.client_secret
 
         # checks if the client id and secret are specified
         if not self.client_id or not self.client_secret:
@@ -42,16 +39,16 @@ def __init__(self, client_id: Optional[str] = None, client_secret: Optional[str]
         # generates a new client access token
         self.access_token: str = self.get_new_access_token()
 
-    def __request_header(self, /, client_id: bool = True, acces_token: bool = True, **kwargs) -> Dict[str, str]:
+    def __request_header(self, /, client_id: bool = True, acces_token: bool = True, **kwargs) -> dict[str, str]:
         """Constructs and returns the request header.
 
         Args:
-            client_id (bool, optional): appends the client id if `True`. Defaults to `True`.
-            acces_token (bool, optional): appends the access token to the header if `True`. Defaults to `True`.
+            client_id (bool, optional): appends the client id if True. Defaults to True.
+            acces_token (bool, optional): appends the access token to the header if True. Defaults to True.
             **kwargs: the additional header attributes.
 
         Returns:
-            Dict[str, str]: the common request header with the specified attributes.
+            dict[str, str]: the common request header with the specified attributes.
         """
         header = {"Accept": "application/vnd.twitchtv.v5+json", **kwargs}
 
@@ -92,7 +89,7 @@ def is_access_token_valid(self, access_token: str) -> bool:
             access_token (str): the client access token.
 
         Returns:
-            bool: `True` if the token is valid, `False` otherwise.
+            bool: True if the token is valid, False otherwise.
         """
         try:
             self.router.request("GET", "validate", {"Authorization": f"OAuth {access_token}"})
@@ -168,7 +165,7 @@ def get_game(self, game_name: str) -> Any:
 
         return result
 
-    def get_game_clips(self, game_id: str, *, limit=100, period=3) -> List[Any]:
+    def get_game_clips(self, game_id: str, *, limit=100, period=3) -> list[Any]:
         """Returns the top clips for the specified game.
 
         See:
@@ -180,7 +177,7 @@ def get_game_clips(self, game_id: str, *, limit=100, period=3) -> List[Any]:
             period (int, optional): the date delta in days. Defaults to 3 days.
 
         Returns:
-            List[Any]: the list of clips if the response was a success.
+            list[Any]: the list of clips if the response was a success.
         """
         # date format is RFC3339 like yyyy-MM-ddTHH:mm:ssZ
         started_at = datetime.datetime.utcnow() - datetime.timedelta(days=max(1, period))
@@ -194,16 +191,16 @@ def get_game_clips(self, game_id: str, *, limit=100, period=3) -> List[Any]:
 
         return self.__parse_clips(header, params)
 
-    def __parse_clips(self, header: Dict[str, str], params: Dict[str, Any], *, pages: int = 10) -> List[Any]:
+    def __parse_clips(self, header: dict[str, str], params: dict[str, Any], *, pages: int = 10) -> list[Any]:
         """Collects and normalizes the clips from the paginated request.
 
         Args:
-            header (Dict[str, str]): the request header.
-            params (Dict[str, Any]): the request parameters.
+            header (dict[str, str]): the request header.
+            params (dict[str, Any]): the request parameters.
             pages (int, optional): the maximum number of pages to request.
 
         Returns:
-            List[Any]: the list of clips.
+            list[Any]: the list of clips.
         """
         # TODO put in signature, in **kwargs or .env ?
         min_views = 50  # the minimum number view count for a clip

compyle/services/youtube.py

@@ -1,18 +1,15 @@
 import os
 import sys
 import webbrowser
+from enum import Enum
 from http.server import BaseHTTPRequestHandler, HTTPServer
-from typing import Dict, List, Optional, Tuple
 from urllib.parse import parse_qs, urlparse
 
 from requests import HTTPError
 
 from compyle.services.common import Routable
 from compyle.settings import YOUTUBE_CONFIG
-from compyle.utils.descriptors import deserialize
-from compyle.utils.types import Enum, Singleton
-
-APP_ROUTES: str = "compyle/services/controllers/routes/youtube.json"
+from compyle.utils.types import Singleton
 
 
 class PrivacyStatus(Enum):
@@ -44,30 +41,27 @@ class YoutubeAPI(Routable):
 
     __metaclass__ = Singleton
 
-    def __init__(
-        self, client_id: Optional[str] = None, client_secret: Optional[str] = None, redirect_uri: Optional[str] = None
-    ):
+    def __init__(self, client_id: str | None = None, client_secret: str | None = None, redirect_uri: str | None = None):
         """Initializes a new instance of the Youtube API client.
 
         Params:
-            client_id (Optional[str]): The client id of the Youtube application. Defaults to None.
-            client_secret (Optional[str]): The client secret of the Youtube application. Defaults to None.
-            redirect_uri (Optional[str]): The redirect uri of the Youtube application. Defaults to None.
+            client_id ( str | None): The client id of the Youtube application. Defaults to None.
+            client_secret ( str | None): The client secret of the Youtube application. Defaults to None.
+            redirect_uri ( str | None): The redirect uri of the Youtube application. Defaults to None.
         """
-        # retrieves the routes description from the JSON file
-        super().__init__(deserialize(APP_ROUTES))
+        super().__init__()
 
         # retrieves the client id and secret either from the parameters or the environment variables
-        self.client_id: Optional[str] = client_id or YOUTUBE_CONFIG.client_id
-        self.client_secret: Optional[str] = client_secret or YOUTUBE_CONFIG.client_secret
+        self.client_id: str | None = client_id or YOUTUBE_CONFIG.client_id
+        self.client_secret: str | None = client_secret or YOUTUBE_CONFIG.client_secret
 
         # checks if the client id and secret are specified
         if not self.client_id or not self.client_secret:
             raise ValueError("The client id and secret must be specified in the environment variables.")
 
         # retrieves the redirect uri and client email either from the parameters or from the environment variables
         self.redirect_uri: str = redirect_uri or YOUTUBE_CONFIG.redirect_uri
-        user_email_address: Optional[str] = YOUTUBE_CONFIG.client_email
+        user_email_address: str | None = YOUTUBE_CONFIG.client_email
 
         # retrieves the authorization code from authentication service
         autorization_code: str = self.authentificate(user_email_address)
@@ -76,7 +70,7 @@ def __init__(
         self.access_token = self.get_access_token(code)
 
     # pylint: disable=line-too-long
-    def authentificate(self, login_hint: Optional[str] = None) -> str:
+    def authentificate(self, login_hint: str | None = None) -> str:
         """Redirect the user to Google's OAuth 2.0 server to initiate the authentication and authorization process.
         Google's OAuth 2.0 server authenticates the user and obtains consent from the user for your application to access the requested scopes.
         The response is sent back to your application using the redirect URL you specified.
@@ -87,7 +81,7 @@ def authentificate(self, login_hint: Optional[str] = None) -> str:
             https://developers.google.com/identity/protocols/oauth2/scopes#youtube for scopes.
 
         Params:
-            login_hint (Optional[str]): the email address of the user to log in. Defaults to None.
+            login_hint ( str | None): the email address of the user to log in. Defaults to None.
 
         Returns:
             str: the authorization code
@@ -184,15 +178,15 @@ def refresh_access_token(self, refresh_token: str):
 
         return self.router.request("POST", "token", header, **params)["access_token"]
 
-    def __request_header(self, /, acces_token: bool = True, **kwargs) -> Dict[str, str]:
+    def __request_header(self, /, acces_token: bool = True, **kwargs) -> dict[str, str]:
         """Constructs and returns the request header.
 
         Args:
-            acces_token (bool, optional): appends the access token to the header if `True`. Defaults to `True`.
+            acces_token (bool, optional): appends the access token to the header if True. Defaults to True.
             **kwargs: the additional header attributes.
 
         Returns:
-            Dict[str, str]: the common request header with the specified attributes.
+            dict[str, str]: the common request header with the specified attributes.
         """
         header = {"Content-Type": "application/json; charset=UTF-8", "Accept": "application/json", **kwargs}
 
@@ -201,7 +195,7 @@ def __request_header(self, /, acces_token: bool = True, **kwargs) -> Dict[str, s
 
         return header
 
-    def get_categories(self, region_code: str = "FR") -> List[Tuple[str, int]]:
+    def get_categories(self, region_code: str = "FR") -> list[tuple[str, int]]:
         """Retrieves the list of categories in the region specified by the region_code in the format ISO 3166-1 alpha-2.
 
         Example:
@@ -212,7 +206,7 @@ def get_categories(self, region_code: str = "FR") -> List[Tuple[str, int]]:
             region_code (str, optional): the code of the region. Defaults to "FR".
 
         Returns:
-            List[Tuple[str, int]]: the list of assignable categories in the specified region.
+            list[tuple[str, int]]: the list of assignable categories in the specified region.
         """
         header = self.__request_header()
         params = {"part": "snippet", "regionCode": region_code}
@@ -240,9 +234,7 @@ def is_category(self, category: str = "22") -> bool:
         return True
 
     # pylint: disable=too-many-arguments
-    def upload_video(
-        self, filename: str, title: str, description: str, category: str, tags: Optional[List[str]] = None
-    ):
+    def upload_video(self, filename: str, title: str, description: str, category: str, tags: list[str] | None = None):
         """Uploads a video to Youtube from the specified file.
 
         See:
@@ -255,7 +247,7 @@ def upload_video(
             title (str): the title of the video.
             description (str): the description of the video.
             category (str): the category id of the video as a string.
-            tags (Optional[List[str]], optional): the tags of the video. Defaults to None.
+            tags (list[str], optional): the tags of the video. Defaults to None.
         """
         with open(filename, "rb") as file:
             body = {
@@ -306,7 +298,7 @@ def upload_video(
 
         exit(0)
 
-    def test(self, filename: str, title: str, description: str, category: str, tags: Optional[List[str]] = None):
+    def test(self, filename: str, title: str, description: str, category: str, tags: list[str] | None = None):
         with open(filename, "rb") as file:
             header = self.__request_header()
             header["Content-Type"] = "video/mp4"