Minor improvements and bugfix

Author: Rajdave69

CHANGELOG.MD

@@ -65,22 +65,22 @@ All notable changes to this project will be documented in this file.
 ### Improved
 - `CircularChecker` class now returns a sorted dict of new circulars
 
-# v1.3.1 - 23/6/2023
+## v1.3.1 - 23/6/2023
 
 ### Fixed
 - Changed the param name of /search to `query` from `title` to match API change
 
-# v1.3.2 - 24/7/2023
+## v1.3.2 - 24/7/2023
 
 ### Fixed
 - Fixed dict key error in `CircularChecker` class.
 
-# v1.3.3 - 8/4/2024
+## v1.3.3 - 8/4/2024
 
-## Improved
+### Improved
 - The package now uses bpsapi.rajtech.me/latest/{category} instead of the params method for list and latest
 
-# v1.4 - 12/8/24
+## v1.4 - 12/8/24
 
 ### Fixed
 - CircularChecker's false positives 
@@ -89,4 +89,13 @@ All notable changes to this project will be documented in this file.
 - Pickle method for cache storage
 
 ### Added
-- mysql method for cache storage
+- mysql method for cache storage
+
+## v1.4.1 - 24/8/24
+
+### Fixed
+- A small bug which caused an error when `amount` wasn't passed in list
+- Fixed type hints having `or` instead of |
+
+### Improved
+- Many little improvements

pybpsapi.py

@@ -1,8 +1,10 @@
+import warnings
 import mysql.connector
 import requests
 import sqlite3
 import os
 
+_min_category_id = 23
 
 class API:
     """Methods which communicate with the API"""
@@ -18,100 +20,99 @@ def __init__(self, url="https://bpsapi.rajtech.me/"):
             raise ConnectionError("Invalid API Response. API says there are no categories.")
 
     # /latest endpoint
-    def latest(self, category: str or int) -> dict | None:
+    def latest(self, category: str | int) -> dict | None:
         """The `/latest` endpoint returns the latest circular from a particular category"""
-        if type(category) is int:
-            category = int(category)
 
-            if not 1 < category < 100:
-                raise ValueError("Invalid category Number")
+        if type(category) is str:
+            if category.isdigit():
+                try:
+                    category = int(category)
+                except ValueError:
+                    warnings.warn(f"Category id: {category} is digit, but cannot be converted to int. "
+                                  f"It will be treated as a category name.")
 
+        # A category id or category name can be passed in
+        if type(category) is int:
+            if not _min_category_id <= category:
+                raise ValueError("Invalid category Number")
         else:
+            # Check with the API's category list
             if category not in self.categories:
                 raise ValueError("Invalid category Name")
 
         request = requests.get(f"{self.url}latest/{category}")
         json = request.json()
 
-        try:
-            json['http_status']
-        except KeyError:
-            raise ConnectionError("Invalid API Response")
-        if json['http_status'] == 200:
-            return json['data']
+        if json.get("data") is None or json.get("http_status") is None:
+            raise ConnectionError("Invalid API Response, it doesn't contain either 'data' or 'http_code'")
+
+        return json['data']
 
     # /list endpoint
-    def list_(self, category: str or int, amount: int = None) -> list | None:
+    def list_(self, category: str | int, amount: int = None) -> list | None:
         """The `/list` endpoint returns a list of circulars from a particular category"""
         if type(category) is int:
-            if not 1 < category < 100:
+            if not _min_category_id <= category:
                 raise ValueError("Invalid category Number")
 
         else:
             if category not in self.categories:
                 raise ValueError("Invalid category Name")
 
-        if amount < 1:
+        if type(amount) is int and amount < 1:
             amount = None
 
         request = requests.get(f"{self.url}list/{category}")
         json = request.json()
 
-        try:
-            json['http_status']
-        except KeyError:
-            raise ConnectionError("Invalid API Response")
-        if json['http_status'] == 200:
-            return json['data'][:amount]
+        if json.get("data") is None or json.get("http_status") is None:
+            raise ConnectionError("Invalid API Response, it doesn't contain either 'data' or 'http_code'")
 
-    def search(self, query: str or int, amount: int = 1) -> dict | None:
+        return json['data'][:amount]
+
+    # /search endpoint
+    def search(self, query: str | int, amount: int = 1) -> dict | None:
         """The `/search` endpoint lets you search for a circular by its name or ID"""
         if query.isdigit() and len(query) == 4:
             query = int(query)
-        elif type(query) != str:
-            raise ValueError("Invalid Query")
+        elif type(query) is not str:
+            raise ValueError("Invalid Query. It isn't string")
 
         params = {'query': query, 'amount': amount}
 
         request = requests.get(self.url + "search", params=params)
         json = request.json()
 
-        try:
-            json['http_status']
-
-        except KeyError:
-            raise ConnectionError("Invalid API Response")
+        if json.get("data") is None or json.get("http_status") is None:
+            raise ConnectionError("Invalid API Response, it doesn't contain either 'data' or 'http_code'")
 
-        if json['http_status'] == 200:
-            return json['data']
+        return json['data']
 
     # /getpng endpoint
     def getpng(self, url: str) -> list | None:
         """The `/getpng` endpoint lets you get the pngs from a circular"""
         if type(url) != str:
-            raise ValueError("Invalid URL")
+            raise ValueError("Invalid URL. It isn't string.")
 
         params = {'url': url}
 
         request = requests.get(self.url + "getpng", params=params)
         json = request.json()
 
-        try:
-            json['http_status']
+        if json.get("data") is None or json.get("http_status") is None:
+            raise ConnectionError("Invalid API Response, it doesn't contain either 'data' or 'http_code'")
 
-        except KeyError:
-            raise ConnectionError("Invalid API Response")
-
-        if json['http_status'] == 200:
-            return json['data']
+        return json['data']
 
 
 class CircularChecker:
-    def __init__(self, category, url: str = "https://bpsapi.rajtech.me/", cache_method='sqlite', **kwargs):
+    def __init__(self, category: str | int, url: str = "https://bpsapi.rajtech.me/", cache_method: str = 'sqlite', **kwargs):
         self.url = url
         self.category = category
+        self.cache_method = cache_method
         self._cache = []
 
+        # Get category names from API
         json = requests.get(self.url + "categories").json()
 
         if json['http_status'] == 200:
@@ -125,18 +126,13 @@ def __init__(self, category, url: str = "https://bpsapi.rajtech.me/", cache_meth
 
         # If category id is passed
         if type(self.category) is int:
-            if not 1 < self.category < 100:
+            if not _min_category_id <= self.category:
                 raise ValueError("Invalid category Number")
-        # If category name is passed
-        else:
+        else:   # If category name is passed
             if self.category not in categories:
                 raise ValueError("Invalid category Name")
 
-        self.cache_method = cache_method
-
-        if cache_method is None:
-            raise ValueError("Invalid Cache Method")
-
+        # For the sqlite cache method
         if self.cache_method == "sqlite":
             try:
                 self.db_name = kwargs['db_name']
@@ -146,12 +142,14 @@ def __init__(self, category, url: str = "https://bpsapi.rajtech.me/", cache_meth
                 raise ValueError(
                     "Invalid Database Parameters. One of db_name, db_path, db_table not passed into kwargs")
 
+            # Create local db if it does not exist
             if not os.path.exists(self.db_path + f"/{self.db_name}.db"):
                 os.mkdir(self.db_path)
 
             self._con = sqlite3.connect(self.db_path + f"/{self.db_name}.db")
             self._cur = self._con.cursor()
 
+        # For the mysql/mariadb cache method
         elif cache_method == "mysql":
             try:
                 self.db_name = kwargs['db_name']
@@ -172,8 +170,9 @@ def __init__(self, category, url: str = "https://bpsapi.rajtech.me/", cache_meth
             self._cur = self._con.cursor(prepared=True)
 
         else:
-            raise Exception("Invalid cache method. Only mysql and sqlite allowed")
+            raise ValueError("Invalid cache method. Only mysql and sqlite allowed")
 
+        # Create a table to cache circulars if it's not there
         self._cur.execute(
             f"""
         CREATE TABLE IF NOT EXISTS  {self.db_table}  (
@@ -187,13 +186,14 @@ def __init__(self, category, url: str = "https://bpsapi.rajtech.me/", cache_meth
         self._con.commit()
 
 
-
+    # Method to retrieve cache from the database
     def get_cache(self) -> list[list] | list:
         self._cur.execute(f"SELECT id, title, link FROM {self.db_table} WHERE category = ?", (self.category,))
         res = self._cur.fetchall()
 
         return res
 
+    # Method to add multiple items to cache
     def _set_cache(self, data):
         # data [ (id, title, link) ]
         query = f"INSERT OR IGNORE INTO {self.db_table} (category, id, title, link) VALUES (?, ?, ?, ?)"
@@ -204,25 +204,26 @@ def _set_cache(self, data):
         self._cur.executemany(query, tuple((self.category, *d) for d in data))
         self._con.commit()
 
-
-    def _add_to_cache(self, id_, title, link):
+    # Method to add a single item to cache
+    def _add_to_cache(self, id_: int, title: str, link: str):
         query = f"INSERT OR IGNORE INTO {self.db_table} (id, title, link) VALUES (?, ?, ?, ?)"
 
         if self.cache_method == 'mysql':
             query = query.replace("OR ", "")
 
         self._cur.execute(query, (self.category, id_, title, link))
 
+    # Method to retrieve circulars from the API and insert into cache
     def _refresh_cache(self):
         request = requests.get(f"{self.url}list/{self.category}")
         json: dict = request.json()
 
-        try:
-            json['http_status']
-        except KeyError:
-            raise ValueError("Invalid API Response")
+        if json.get("data") is None or json.get("http_status") is None:
+            raise ConnectionError("Invalid API Response, it doesn't contain either 'data' or 'http_code'")
 
         self._cur.execute(f"SELECT id FROM {self.db_table} WHERE category = ?", (self.category,))
+
+        # ((1234,), (4567,), ...) -> ('1234', '4567')
         cached_ids: list = self._cur.fetchall()
         cached_ids: tuple[str, ...] = tuple([str(i[0]) for i in cached_ids])
 
@@ -231,22 +232,24 @@ def _refresh_cache(self):
                 [
                     (i['id'], i['title'], i['link'])
                     for i in json['data']
-                    if i['id'] not in cached_ids
+                    if i['id'] not in cached_ids    # Add only new circulars to the database
                 ]
             )
 
-    def check(self) -> list[dict] | list[None]:
+    # Method to check for new circular(s)
+    def check(self) -> list[dict] | list:
+        # First get cached circulars and store them in a variable 'cached_circular_ids'
+        # Then refresh cache and get the new list of circulars, and then compare and find new ones.
         self._cur.execute(f"SELECT id FROM {self.db_table} WHERE category = ?", (self.category,))
 
         cached_circular_ids = self._cur.fetchall()
-        cached_circular_ids = [i[0] for i in cached_circular_ids]
+        cached_circular_ids = [i[0] for i in cached_circular_ids]   # [(id, title, link)]
 
         self._refresh_cache()
-        new_circular_list = self.get_cache()
-        # data [(id, title, link)]
-
-        if len(new_circular_list) != len(cached_circular_ids):
+        new_circular_list = self.get_cache() #
 
+        # If there are new circulars
+        if len(new_circular_list) > len(cached_circular_ids):
             new_circular_objects = [i for i in new_circular_list if i[0] not in cached_circular_ids]
 
             # (id, title, link) -> {'id': id, 'title': title, 'link': link}
@@ -263,45 +266,60 @@ def check(self) -> list[dict] | list[None]:
             new_circular_objects.sort(key=lambda x: x['id'])
             return new_circular_objects
 
-        else:
-            return []
+        return []
+
+    # Close connections when object is deleted
+    def __del__(self):
+        if hasattr(self, '_con'):
+            self._con.close()
 
 
 class CircularCheckerGroup:
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *circular_checkers: CircularChecker, **kwargs):
         self._checkers = []
 
-        for arg in args:
-            if type(arg) is not CircularChecker:
+        # Add each checker to self._checkers
+        for checker in circular_checkers:
+            if type(checker) is not CircularChecker:
                 raise ValueError("Invalid CircularChecker Object")
-            self._checkers.append(arg)
+            self._checkers.append(checker)
 
         if bool(kwargs.get("debug")):
             self.checkers = self._checkers
 
-    def add(self, checker: CircularChecker, *args: CircularChecker):
+    # Method to add a circular checker to this group
+    def add(self, checker: CircularChecker, *circular_checkers: CircularChecker):
         self._checkers.append(checker)
-        for arg in args:
-            if type(arg) is not CircularChecker:
+
+        for checker in circular_checkers:
+            if type(checker) is not CircularChecker:
                 raise ValueError("Invalid CircularChecker Object")
-            self._checkers.append(arg)
+            self._checkers.append(checker)
 
+    # Method to create a circular checker and add it to the group
     def create(self, category, url: str = "https://bpsapi.rajtech.me/", cache_method=None, **kwargs):
         checker = CircularChecker(category, url, cache_method, **kwargs)
         self._checkers.append(checker)
 
+    # Method to check for new circulars in each one of the checkers
     def check(self) -> dict[list[dict], ...] | dict:
         return_dict = {}
         for checker in self._checkers:
             return_dict[checker.category] = checker.check()
         return return_dict
 
+    # Method to refresh (sync) cache from API
     def refresh_cache(self):
         for checker in self._checkers:
             checker.refresh_cache()
 
+    # Method to get the cache of all checkers
     def get_cache(self) -> dict[list[list]] | dict:
         return_dict = {}
         for checker in self._checkers:
             return_dict[checker.category] = checker.get_cache()
         return return_dict
+
+    def __del__(self):
+        for checker in self._checkers:
+            del checker