Merge branch 'release/port-bans-and-calendar-methods'

setup.py

@@ -11,7 +11,7 @@
 
 setup(
     name='shikithon',
-    version="0.0.2",
+    version="0.0.3",
     license='MIT',
     description='Yet another Python wrapper for Shikimori API',
     long_description=long_description,

shikithon/api.py

@@ -10,6 +10,8 @@
 from .enums.Anime import *
 from .models.Achievement import Achievement
 from .models.Anime import Anime
+from .models.Ban import Ban
+from .models.CalendarEvent import CalendarEvent
 from .models.Creator import Creator
 from .models.FranchiseTree import FranchiseTree
 from .models.Link import Link
@@ -23,42 +25,42 @@ class API:
     def __init__(self, api_config: Dict[str, str]):
         self.endpoints: APIEndpoints = APIEndpoints()
         self.session: Session = Session()
-        self.__init_api_vars(api_config)
-        self.__api_vars_validation()
+        self.init_api_vars(api_config)
+        self.api_vars_validation()
 
-    def __get(self, url: str, headers: Dict[str, str] = {}, query: Dict[str, str] = {}, data: Dict[str, str] = {}):
+    def get(self, url: str, headers: Dict[str, str] = {}, query: Dict[str, str] = {}, data: Dict[str, str] = {}):
         response = self.session.get(url, headers=headers, params=query, data=data)
         if response.status_code == 401:
-            self.__update_tokens(self.__get_tokens_from_api(refresh_tokens=True))
-            return self.__get(url, headers, query, data)
+            self.update_tokens(self.get_tokens_from_api(refresh_tokens=True))
+            return self.get(url, headers, query, data)
         return response.json()
 
-    def __post(self, url: str, headers: Dict[str, str] = {}, query: Dict[str, str] = {}, data: Dict[str, str] = {}):
+    def post(self, url: str, headers: Dict[str, str] = {}, query: Dict[str, str] = {}, data: Dict[str, str] = {}):
         response = self.session.post(url, headers=headers, params=query, data=data)
         if response.status_code == 401:
-            self.__update_tokens(self.__get_tokens_from_api(refresh_tokens=True))
-            return self.__post(url, headers, query, data)
+            self.update_tokens(self.get_tokens_from_api(refresh_tokens=True))
+            return self.post(url, headers, query, data)
         return response.json()
 
-    def __put(self, url: str, headers: Dict[str, str] = {}, query: Dict[str, str] = {}, data: Dict[str, str] = {}):
+    def put(self, url: str, headers: Dict[str, str] = {}, query: Dict[str, str] = {}, data: Dict[str, str] = {}):
         response = self.session.put(url, headers=headers, params=query, data=data)
         if response.status_code == 401:
-            self.__update_tokens(self.__get_tokens_from_api(refresh_tokens=True))
-            return self.__put(url, headers, query, data)
+            self.update_tokens(self.get_tokens_from_api(refresh_tokens=True))
+            return self.put(url, headers, query, data)
         return response.json()
 
-    def __patch(self, url: str, headers: Dict[str, str] = {}, query: Dict[str, str] = {}, data: Dict[str, str] = {}):
+    def patch(self, url: str, headers: Dict[str, str] = {}, query: Dict[str, str] = {}, data: Dict[str, str] = {}):
         response = self.session.patch(url, headers=headers, params=query, data=data)
         if response.status_code == 401:
-            self.__update_tokens(self.__get_tokens_from_api(refresh_tokens=True))
-            return self.__patch(url, headers, query, data)
+            self.update_tokens(self.get_tokens_from_api(refresh_tokens=True))
+            return self.patch(url, headers, query, data)
         return response.json()
 
-    def __delete(self, url: str, headers: Dict[str, str] = {}, query: Dict[str, str] = {}, data: Dict[str, str] = {}):
+    def delete(self, url: str, headers: Dict[str, str] = {}, query: Dict[str, str] = {}, data: Dict[str, str] = {}):
         response = self.session.delete(url, headers=headers, params=query, data=data)
         if response.status_code == 401:
-            self.__update_tokens(self.__get_tokens_from_api(refresh_tokens=True))
-            return self.__delete(url, headers, query, data)
+            self.update_tokens(self.get_tokens_from_api(refresh_tokens=True))
+            return self.delete(url, headers, query, data)
         return response.json()
 
     def get_api_config_dict(self) -> Dict[str, str]:
@@ -75,15 +77,15 @@ def get_api_config_dict(self) -> Dict[str, str]:
         }
 
     def set_api_config_dict(self, api_config: Dict[str, str]):
-        self.__init_api_vars(api_config)
-        self.__api_vars_validation()
+        self.init_api_vars(api_config)
+        self.api_vars_validation()
 
-    def __export_api_config(self, file_name: str = "new_config.json"):
+    def export_api_config(self, file_name: str = "new_config.json"):
         current_config: Dict[str, str] = self.get_api_config_dict()
         with open(file_name, "w", encoding="utf-8") as config_file:
             json.dump(current_config, config_file, indent=4, separators=(", ", ": "))
 
-    def __init_api_vars(self, api_config):
+    def init_api_vars(self, api_config):
         try:
             self.app_name: str = api_config["app_name"]
             self.client_id: str = api_config["client_id"]
@@ -94,9 +96,9 @@ def __init_api_vars(self, api_config):
             self.access_token: str = api_config["access_token"]
             self.refresh_token: str = api_config["refresh_token"]
         except KeyError:
-            self.__raise_config_mismatch(api_config)
+            self.raise_config_mismatch(api_config)
 
-    def __api_vars_validation(self):
+    def api_vars_validation(self):
         if not self.app_name:
             raise MissingConfigData("To use the Shikimori API correctly, you need to pass the application name")
 
@@ -107,24 +109,24 @@ def __api_vars_validation(self):
 
         if not self.auth_code:
             oauth_url_data: Union[str, List[str]] = self.endpoints.get_filled_oauth_url(self.client_id, self.redirect_uri, self.scopes)
-            exception_msg: str = self.__parse_oauth_url_data(oauth_url_data)
+            exception_msg: str = self.parse_oauth_url_data(oauth_url_data)
             raise MissingConfigData(exception_msg)
 
         if not self.access_token or not self.refresh_token:
-            self.__update_tokens(self.__get_tokens_from_api())
+            self.update_tokens(self.get_tokens_from_api())
 
         self.session.headers.update({"Authorization": f"Bearer {self.access_token}"})
 
-    def __raise_config_mismatch(self, api_config):
+    def raise_config_mismatch(self, api_config):
         configs_keys: Tuple[List[str], List[str]] = list(api_config.keys()), list(APIHelpers.get_blank_config().keys())
         missing_keys_str: str = ", ".join(APIHelpers.get_missing_keys_list(configs_keys))
         raise MissingConfigData(f"It is impossible to initialize an object without missing variables. Here is the list of missing: {missing_keys_str}")
 
-    def __update_tokens(self, tokens_tuple: Tuple[str, str]):
+    def update_tokens(self, tokens_tuple: Tuple[str, str]):
         self.access_token, self.refresh_token = tokens_tuple
-        self.__export_api_config()
+        self.export_api_config()
 
-    def __get_tokens_from_api(self, refresh_tokens: bool = False) -> Tuple[str, str]:
+    def get_tokens_from_api(self, refresh_tokens: bool = False) -> Tuple[str, str]:
         token_url = self.endpoints.get_token_url()
         data = {
             "client_id": self.client_id,
@@ -139,15 +141,15 @@ def __get_tokens_from_api(self, refresh_tokens: bool = False) -> Tuple[str, str]
             data["code"] = self.auth_code
             data["redirect_uri"] = self.redirect_uri
 
-        oauth_json = self.__post(url=token_url, data=data)
+        oauth_json = self.post(url=token_url, data=data)
 
         try:
             return oauth_json["access_token"], oauth_json["refresh_token"]
         except KeyError:
             error_info = json.dumps(oauth_json)
             raise MissingConfigData(f"An error occurred while receiving tokens, here is the information from the response: {error_info}")
 
-    def __parse_oauth_url_data(self, oauth_url_data):
+    def parse_oauth_url_data(self, oauth_url_data):
         if isinstance(oauth_url_data, list):
             missing_variables_str = ", ".join(oauth_url_data)
             return "The authorization code and some related information necessary to obtain it are missing: " \
@@ -160,7 +162,7 @@ def get_achievements(self, user_id: int) -> List[Achievement]:
         query: Dict[str, str] = {
             "user_id": str(user_id)
         }
-        res: List[Dict[str, Any]] = self.__get(url=self.endpoints.get_achievements_url(), query=query)
+        res: List[Dict[str, Any]] = self.get(url=self.endpoints.get_achievements_url(), query=query)
 
         return [Achievement(**data) for data in res]
 
@@ -221,7 +223,7 @@ def get_list_of_animes(self, page: int = 1, limit: int = 1, order: Order = Order
             "exclude_ids": ",".join([str(id) for id in exclude_ids]),
             "search": search
         }
-        res: List[Dict[str, Any]] = self.__get(url=self.endpoints.get_animes_url(), query=query)
+        res: List[Dict[str, Any]] = self.get(url=self.endpoints.get_animes_url(), query=query)
         return [Anime(**anime) for anime in res]
 
     def get_anime(self, anime_id: int) -> Anime:
@@ -234,7 +236,7 @@ def get_anime(self, anime_id: int) -> Anime:
             Returns:
                 Anime: info about anime
         """
-        res: Dict[str, Any] = self.__get(url=self.endpoints.get_anime_url(anime_id))
+        res: Dict[str, Any] = self.get(url=self.endpoints.get_anime_url(anime_id))
         return Anime(**res)
 
     def get_anime_creators(self, anime_id: int) -> List[Creator]:
@@ -247,7 +249,7 @@ def get_anime_creators(self, anime_id: int) -> List[Creator]:
             Returns:
                 List[Creator]: list of anime creators
         """
-        res: List[Dict[str, Any]] = self.__get(url=self.endpoints.get_anime_roles_url(anime_id))
+        res: List[Dict[str, Any]] = self.get(url=self.endpoints.get_anime_roles_url(anime_id))
         return [Creator(**creator) for creator in res]
 
     def get_list_of_similar_animes(self, anime_id: int) -> List[Anime]:
@@ -260,7 +262,7 @@ def get_list_of_similar_animes(self, anime_id: int) -> List[Anime]:
             Returns:
                 List[Anime]: list of similar animes
         """
-        res: List[Dict[str, Any]] = self.__get(url=self.endpoints.get_similar_animes_url(anime_id))
+        res: List[Dict[str, Any]] = self.get(url=self.endpoints.get_similar_animes_url(anime_id))
         return [Anime(**anime) for anime in res]
 
     def get_anime_related_content(self, anime_id: int) -> List[Relation]:
@@ -273,7 +275,7 @@ def get_anime_related_content(self, anime_id: int) -> List[Relation]:
             Returns:
                 List[Relation]: list of anime related content
         """
-        res: List[Dict[str, Any]] = self.__get(url=self.endpoints.get_anime_related_content_url(anime_id))
+        res: List[Dict[str, Any]] = self.get(url=self.endpoints.get_anime_related_content_url(anime_id))
         return [Relation(**relation) for relation in res]
 
     def get_anime_screenshots(self, anime_id: int) -> List[Screenshot]:
@@ -286,7 +288,7 @@ def get_anime_screenshots(self, anime_id: int) -> List[Screenshot]:
             Returns:
                 List[Screenshot]: list of anime screenshots links
         """
-        res: List[Dict[str, Any]] = self.__get(url=self.endpoints.get_anime_screenshots_url(anime_id))
+        res: List[Dict[str, Any]] = self.get(url=self.endpoints.get_anime_screenshots_url(anime_id))
         return [Screenshot(**screenshot) for screenshot in res]
 
     def get_anime_franchise_tree(self, anime_id: int) -> FranchiseTree:
@@ -299,7 +301,7 @@ def get_anime_franchise_tree(self, anime_id: int) -> FranchiseTree:
             Returns:
                 FranchiseTree: franchise tree of anime
         """
-        res: Dict[str, Any] = self.__get(url=self.endpoints.get_anime_franchise_tree_url(anime_id))
+        res: Dict[str, Any] = self.get(url=self.endpoints.get_anime_franchise_tree_url(anime_id))
         return FranchiseTree(**res)
 
     def get_anime_external_links(self, anime_id: int) -> List[Link]:
@@ -312,7 +314,7 @@ def get_anime_external_links(self, anime_id: int) -> List[Link]:
             Returns:
                 List[Link]: list of anime links
         """
-        res: List[Dict[str, Any]] = self.__get(url=self.endpoints.get_anime_external_links_url(anime_id))
+        res: List[Dict[str, Any]] = self.get(url=self.endpoints.get_anime_external_links_url(anime_id))
         return [Link(**link) for link in res]
 
     def get_anime_topics(self, anime_id: int, page: int = 1, limit: int = 1, kind: Status = Status.EPISODE, episode: Union[int, str] = "") -> List[Topic]:
@@ -343,11 +345,53 @@ def get_anime_topics(self, anime_id: int, page: int = 1, limit: int = 1, kind: S
             "kind": kind.value,
             "episode": str(episode)
         }
-        res: List[Dict[str, Any]] = self.__get(url=self.endpoints.get_anime_topics_url(anime_id), query=query)
+        res: List[Dict[str, Any]] = self.get(url=self.endpoints.get_anime_topics_url(anime_id), query=query)
         return [Topic(**topic) for topic in res]
 
+    def get_bans_list(self, page: int = 1, limit: int = 1) -> list[Ban]:
+        """
+            Returns list of recent bans on Shikimori.
+
+            Current API method returns `limit + 1` elements, if API has next page.
+
+            Parameters:
+                page (int): Number of page (Defaults to 1)
+                limit (int): Number of results (Default to 1)
+
+            Returns:
+                list[Ban]: list of recent bans
+        """
+        if page < 1 or page > 100000:
+            page = 1
+
+        if limit < 1 or limit > 30:
+            limit = 1
+
+        query: Dict[str, str] = {
+            "page": str(page),
+            "limit": str(limit)
+        }
+        res: List[Dict[str, Any]] = self.get(url=self.endpoints.get_bans_list_url(), query=query)
+        return [Ban(**ban) for ban in res]
+
+    def get_current_calendar(self, censored: Censorship = Censorship.CENSORED) -> list[CalendarEvent]:
+        """
+            Returns current calendar events.
+
+            Parameters:
+                censored (Censorship): Status of censorship (Defaults to Censorship.CENSORED)
+
+            Returns:
+                list[CalendarEvent]: list of calendar events
+        """
+        query: Dict[str, str] = {
+            "censored": censored.value
+        }
+        res: List[Dict[str, Any]] = self.get(url=self.endpoints.get_calendar_url(), query=query)
+        return [CalendarEvent(**calendar_event) for calendar_event in res]
+
     def get_current_user(self) -> User:
-        res: Dict[str, Any] = self.__get(url=self.endpoints.get_whoami_url())
+        res: Dict[str, Any] = self.get(url=self.endpoints.get_whoami_url())
         return User(**res)
 
 class APIHelpers:
@@ -439,6 +483,14 @@ def get_anime_external_links_url(self, anime_id: int) -> str:
     def get_anime_topics_url(self, anime_id: int) -> str:
         return f"{self.base_url}/animes/{anime_id}/topics"
 
+    # Bans
+    def get_bans_list_url(self) -> str:
+        return f"{self.base_url}/bans"
+
+    # Calendar
+    def get_calendar_url(self) -> str:
+        return f"{self.base_url}/calendar"
+
     # Users
     def get_whoami_url(self) -> str:
         return f"{self.base_url}/users/whoami"

shikithon/models/Ban.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from typing import Optional
 
 from pydantic import BaseModel
 
@@ -9,7 +10,7 @@
 class Ban(BaseModel):
     id: int
     user_id: int
-    comment: Comment
+    comment: Optional[Comment]
     moderator_id: int
     reason: str
     created_at: datetime

shikithon/models/Calendar.py → shikithon/models/CalendarEvent.py

@@ -6,7 +6,7 @@
 from .Anime import Anime
 
 
-class Calendar(BaseModel):
+class CalendarEvent(BaseModel):
     next_episode: int
     next_episode_at: datetime
     duration: Optional[int]

shikithon/models/Comment.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from typing import Optional
 
 from pydantic import BaseModel
 
@@ -11,5 +12,5 @@ class Comment(BaseModel):
     user_id: int
     created_at: datetime
     updated_at: datetime
-    is_summary: bool
+    is_summary: Optional[bool]
     is_offtopic: bool

shikithon/models/User.py

@@ -14,7 +14,7 @@ class User(BaseModel):
     avatar: str
     image: UserImage
     last_online_at: datetime
-    url: str
+    url: Optional[str]
     name: Optional[str]
     sex: Optional[str]
     full_years: Optional[int]