Merge pull request #24 from bellingcat/4.0/beta

4.0/beta

Author: ritchiemwewa

src/octosuite/app.py

@@ -1,5 +1,6 @@
 import sys
 
+from .core.cache import cache
 from .lib import console, __pkg__, __version__
 from .tui.menus import Menus
 
@@ -11,3 +12,5 @@ def start():
         menu.main()
     except KeyboardInterrupt:
         sys.exit()
+    finally:
+        cache.clear()

src/octosuite/core/cache.py

@@ -0,0 +1,44 @@
+import hashlib
+import json
+
+__all__ = ["cache"]
+
+
+class ResponseCache:
+    """Simple in-memory cache for API responses."""
+
+    def __init__(self):
+        self._cache = {}
+
+    @staticmethod
+    def _generate_key(url: str, params: dict = None) -> str:
+        """Generate a unique cache key from URL and params."""
+        cache_data = f"{url}:{json.dumps(params or {}, sort_keys=True)}"
+        return hashlib.md5(cache_data.encode()).hexdigest()
+
+    def get(self, url: str, params: dict = None):
+        """Get cached response."""
+        key = self._generate_key(url, params)
+        return self._cache.get(key)
+
+    def set(self, url: str, data, params: dict = None):
+        """Cache a response."""
+        key = self._generate_key(url, params)
+        self._cache[key] = data
+
+    def clear(self):
+        """Clear all cached responses."""
+        self._cache.clear()
+
+    def remove(self, url: str, params: dict = None):
+        """Remove a specific cached response."""
+        key = self._generate_key(url, params)
+        self._cache.pop(key, None)
+
+
+cache = ResponseCache()
+
+
+def clear_cache():
+    """Clear all cached API responses."""
+    cache.clear()

src/octosuite/core/github.py

@@ -5,6 +5,7 @@
 import requests
 from requests import Response
 
+from .cache import cache
 from ..lib import __version__
 
 BASE_URL = "https://api.github.com"
@@ -22,10 +23,20 @@ def __init__(
         ),
     ):
         self.user_agent = user_agent
+        self.cache = cache
 
     def get(
-        self, url: str, params: t.Optional[dict] = None, return_response: bool = False
+        self,
+        url: str,
+        params: t.Optional[dict] = None,
+        return_response: bool = False,
+        use_cache: bool = True,
     ) -> t.Union[dict, list, Response]:
+        if use_cache and not return_response:
+            cached = self.cache.get(url, params)
+            if cached is not None:
+                return cached
+
         response = requests.get(
             url=url, params=params, headers={"User-Agent": self.user_agent}
         )
@@ -34,33 +45,51 @@ def get(
             return response
 
         if response.status_code == 200:
-            return self._sanitise_response(response=response.json())
+            sanitised = self.sanitise_response(response=response.json())
+
+            # Cache the successful response
+            if use_cache:
+                self.cache.set(url, sanitised, params)
+
+            return sanitised
+
         return []
 
     def is_valid_entity(
-        self, entity_type: t.Literal["user", "org", "repo"], **kwargs
+        self, _type: t.Literal["user", "org", "repo"], **kwargs
     ) -> bool:
         """Validate if a GitHub entity exists."""
         try:
-            if entity_type == "user":
-                url = f"https://api.github.com/users/{kwargs.get('username')}"
-            elif entity_type == "org":
-                url = f"https://api.github.com/orgs/{kwargs.get('username')}"
-            elif entity_type == "repo":
-                url = f"https://api.github.com/repos/{kwargs.get('repo_owner')}/{kwargs.get('repo_name')}"
-            else:
+            type_map = {
+                "user": f"https://api.github.com/users/{kwargs.get('username')}",
+                "org": f"https://api.github.com/orgs/{kwargs.get('username')}",
+                "repo": f"https://api.github.com/repos/{kwargs.get('repo_owner')}/{kwargs.get('repo_name')}",
+            }
+
+            url = type_map[_type]
+
+            # Check cache first
+            cached = self.cache.get(url)
+            if cached is not None:
+                return True  # If cached, entity exists
+
+            response = requests.get(url=url, headers={"User-Agent": self.user_agent})
+
+            # Only cache if entity exists (status 200)
+            if response.status_code == 200:
+                sanitised = self.sanitise_response(response.json())
+                self.cache.set(url, sanitised)
                 return True
 
-            response = self.get(url=url, return_response=True)
-            return response.status_code == 200
+            return False
         except requests.RequestException:
             return False
 
-    def _sanitise_response(self, response: t.Union[dict, list]) -> t.Union[dict, list]:
+    def sanitise_response(self, response: t.Union[dict, list]) -> t.Union[dict, list]:
         pattern = re.compile(r"https://api\.github\.com")
 
         if isinstance(response, list):
-            return [self._sanitise_response(response=item) for item in response]
+            return [self.sanitise_response(response=item) for item in response]
 
         if isinstance(response, dict):
             keys_to_remove = [
@@ -74,6 +103,6 @@ def _sanitise_response(self, response: t.Union[dict, list]) -> t.Union[dict, lis
             # Recursively clean nested dicts/lists
             for key, value in response.items():
                 if isinstance(value, (dict, list)):
-                    response[key] = self._sanitise_response(response=value)
+                    response[key] = self.sanitise_response(response=value)
 
         return response

src/octosuite/core/models.py

@@ -14,13 +14,27 @@ def __init__(self, source: str):
         self.endpoint = None
         self.source = source
 
-    def exists(self) -> bool:
+    def exists(self) -> tuple[bool, dict]:
         """Check if the entity exists on GitHub."""
+        # Check cache first
+        cached = github.cache.get(self.endpoint)
+        if cached is not None:
+            return True, cached
+
         try:
             response = github.get(url=self.endpoint, return_response=True)
-            return response.status_code == 200
-        except exceptions.RequestException as err:
-            return False
+
+            if response.status_code == 200:
+                data = response.json()
+                # Sanitise the data
+                sanitised = github.sanitise_response(data)
+                # Cache the sanitised response
+                github.cache.set(self.endpoint, sanitised)
+                return True, sanitised
+
+            return False, response.json()
+        except exceptions.RequestException:
+            return False, {}
 
 
 class User(GitHubEntity):

src/octosuite/lib.py

@@ -190,7 +190,7 @@ def check_updates():
     with console.status("[dim]Checking for updates...[/dim]") as status:
         checker = UpdateChecker()
         result = checker.check(__pkg__, __version__)
-        if result:
+        if result is not None:
             status.stop()
             message_dialog(title="Update Available", text=result).run()
         else:

src/octosuite/tui/dialogs.py

@@ -29,17 +29,28 @@ class Dialogs:
     def __init__(self): ...
 
     @staticmethod
-    def quit() -> bool:
+    def _boolean(title: str, text: str) -> bool:
         try:
             result = button_dialog(
-                title="Quit",
-                text="This will close the session, continue?",
+                title=title,
+                text=text,
                 buttons=[("Yes", True), ("No", False)],
             ).run()
             return result if result is not None else False
         except KeyboardInterrupt:
             return True
 
+    def quit(self) -> bool:
+        return self._boolean(
+            title="Quit", text="This will close the session, continue?"
+        )
+
+    def clear_cache(self) -> bool:
+        return self._boolean(
+            title="Clear Cache",
+            text="This will clear all octosuite caches, continue?",
+        )
+
     @staticmethod
     def license():
         message_dialog(title="MIT License", text=LICENSE_NOTICE).run()