Allow response caching

This commit is contained in:
Ritchie Mwewa
2026-01-03 22:51:52 +02:00
parent 66abbcbef6
commit 38a9a4b6b3
7 changed files with 329 additions and 165 deletions

View File

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

View File

@@ -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()

View File

@@ -5,6 +5,7 @@ import typing as t
import requests import requests
from requests import Response from requests import Response
from .cache import cache
from ..lib import __version__ from ..lib import __version__
BASE_URL = "https://api.github.com" BASE_URL = "https://api.github.com"
@@ -22,10 +23,20 @@ class GitHub:
), ),
): ):
self.user_agent = user_agent self.user_agent = user_agent
self.cache = cache
def get( 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]: ) -> 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( response = requests.get(
url=url, params=params, headers={"User-Agent": self.user_agent} url=url, params=params, headers={"User-Agent": self.user_agent}
) )
@@ -34,25 +45,43 @@ class GitHub:
return response return response
if response.status_code == 200: 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 [] return []
def is_valid_entity( def is_valid_entity(
self, entity_type: t.Literal["user", "org", "repo"], **kwargs self, _type: t.Literal["user", "org", "repo"], **kwargs
) -> bool: ) -> bool:
"""Validate if a GitHub entity exists.""" """Validate if a GitHub entity exists."""
try: try:
if entity_type == "user": type_map = {
url = f"https://api.github.com/users/{kwargs.get('username')}" "user": f"https://api.github.com/users/{kwargs.get('username')}",
elif entity_type == "org": "org": f"https://api.github.com/orgs/{kwargs.get('username')}",
url = f"https://api.github.com/orgs/{kwargs.get('username')}" "repo": f"https://api.github.com/repos/{kwargs.get('repo_owner')}/{kwargs.get('repo_name')}",
elif entity_type == "repo": }
url = f"https://api.github.com/repos/{kwargs.get('repo_owner')}/{kwargs.get('repo_name')}"
else: 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 return True
response = self.get(url=url, return_response=True) return False
return response.status_code == 200
except requests.RequestException: except requests.RequestException:
return False return False

View File

@@ -1,5 +1,3 @@
from requests import exceptions
from .github import GitHub, BASE_URL from .github import GitHub, BASE_URL
github = GitHub() github = GitHub()
@@ -14,13 +12,30 @@ class GitHubEntity:
self.endpoint = None self.endpoint = None
self.source = source self.source = source
def exists(self) -> bool: def exists(self) -> tuple[bool, dict]:
"""Check if the entity exists on GitHub.""" """Check if the entity exists on GitHub."""
try: _type = None
response = github.get(url=self.endpoint, return_response=True) kwargs = {}
return response.status_code == 200
except exceptions.RequestException as err: if isinstance(self, User):
return False _type = "user"
kwargs = {"username": self.name}
elif isinstance(self, Org):
_type = "org"
kwargs = {"username": self.name}
elif isinstance(self, Repo):
_type = "repo"
kwargs = {"repo_owner": self.owner, "repo_name": self.name}
# Use is_valid_entity which handles caching and sanitization
exists = github.is_valid_entity(_type=_type, **kwargs)
if exists:
# Get the cached sanitized data
cached = github.cache.get(self.endpoint)
return True, cached if cached else {}
return False, {}
class User(GitHubEntity): class User(GitHubEntity):

View File

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

View File

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

View File

@@ -7,6 +7,7 @@ from rich.status import Status
from .dialogs import Dialogs from .dialogs import Dialogs
from .prompts import Prompts from .prompts import Prompts
from ..core.cache import cache
from ..core.models import User, Repo, Org, Search from ..core.models import User, Repo, Org, Search
from ..lib import check_updates, preview_response, export_response, set_menu_title from ..lib import check_updates, preview_response, export_response, set_menu_title
from ..lib import console, clear_screen, ascii_banner from ..lib import console, clear_screen, ascii_banner
@@ -28,8 +29,10 @@ prompts = Prompts()
__all__ = ["Menus"] __all__ = ["Menus"]
class Menus: class BaseMenu:
def __init__(self): """Base class with common menu functionality."""
def __init__(self, main_menu: t.Callable):
# Define which methods require pagination # Define which methods require pagination
self.paginated_methods = { self.paginated_methods = {
"repos", "repos",
@@ -64,14 +67,87 @@ class Menus:
# Search methods (all require pagination) # Search methods (all require pagination)
self.search_methods = {"repos", "users", "commits", "issues", "topics"} self.search_methods = {"repos", "users", "commits", "issues", "topics"}
self.mode_handlers = { self.main_menu = main_menu
"user": self.user,
"repo": self.repo,
"org": self.org,
"search": self.search,
}
def _execute_selection(self, status: Status, **kwargs): @staticmethod
def target_validator(
identifier: str,
instance: t.Union[User, Repo, Org],
callback: t.Callable,
*callback_args,
) -> bool:
"""Validate if an entity exists on GitHub.
Args:
identifier: Name/identifier of the entity
instance: The instance with an exists() method (User, Repo, or Org)
callback: Function to call if validation fails
*callback_args: Arguments to pass to callback
Returns:
bool: True if entity exists, False otherwise
"""
if isinstance(instance, User):
target_type = "user"
elif isinstance(instance, Repo):
target_type = "repo"
else:
target_type = "org"
with Status(
f"[dim]Validating {target_type}'s ({identifier}) existence...[/dim]",
console=console,
) as status:
exists, response = instance.exists()
if exists:
ascii_banner(text=identifier)
console.print(
f"[bold][green]✔[/green] {target_type.capitalize()} ({identifier}) exists on GitHub[/bold]"
)
return True
else:
status.stop()
if response and "message" in response:
console.print(
f"[bold][yellow]✘[/yellow] {response['message']}[/bold]"
)
console.input(" Press [bold]ENTER[/bold] to continue ...")
callback(*callback_args)
return False
def execute_and_handle_response(
self,
instance: t.Union[User, Repo, Org, Search],
method_name: str,
target_type: t.Literal["user", "repo", "org"],
source: str,
):
"""Execute a method and handle its response.
Args:
instance: The instance to execute the method on
method_name: Name of the method to execute
target_type: Type of entity ("user", "repo", "org")
source: Source identifier for response handling
"""
valid_methods = self.paginated_methods | self.non_paginated_methods
if method_name in valid_methods:
with Status(
status=f"[dim]Initialising {target_type} {method_name}...[/dim]",
console=console,
) as status:
data = self.execute_selection(
source=source,
instance=instance,
method_name=method_name,
status=status,
)
status.stop()
self.response_handler(data=data, data_type=method_name, source=source)
def execute_selection(self, status: Status, **kwargs):
"""Execute a method on an instance, prompting for pagination if needed.""" """Execute a method on an instance, prompting for pagination if needed."""
instance = kwargs.get("instance") instance = kwargs.get("instance")
method_name = kwargs.get("method_name") method_name = kwargs.get("method_name")
@@ -87,22 +163,7 @@ class Menus:
status.update(f"[dim]Getting {method_name} from {source}...[/dim]") status.update(f"[dim]Getting {method_name} from {source}...[/dim]")
return method(**params) return method(**params)
def _navigation(self, option: str, callback: t.Callable, *callback_args): def response_handler(self, data: t.Union[dict, list], data_type: str, source: str):
"""Handle navigation options (back, quit, change settings)."""
navigation_handlers = {
"back": lambda: self.main(),
"quit": lambda: (
sys.exit() if dialogs.quit() else callback(*callback_args)
),
}
handler = navigation_handlers.get(option)
if handler:
handler()
return True
return False
def response_handling(self, data: t.Union[dict, list], data_type: str, source: str):
"""Export data to file in user-selected format(s).""" """Export data to file in user-selected format(s)."""
preview_response(data=data, source=source, _type=data_type) preview_response(data=data, source=source, _type=data_type)
@@ -140,8 +201,10 @@ class Menus:
if dialogs.quit(): if dialogs.quit():
sys.exit() sys.exit()
else: else:
self.response_handling( self.response_handler(
data=data, data_type=data_type, source=source data=data,
data_type=data_type,
source=source,
) )
return return
@@ -177,6 +240,33 @@ class Menus:
except KeyboardInterrupt: except KeyboardInterrupt:
print("\nExport cancelled") print("\nExport cancelled")
def navigation_handler(self, option: str, callback: t.Callable, *callback_args):
"""Handle navigation options (back, quit, change settings)."""
navigation_handlers = {
"back": lambda: self.main_menu(),
"quit": lambda: (
sys.exit() if dialogs.quit() else callback(*callback_args)
),
}
handler = navigation_handlers.get(option)
if handler:
handler()
return True
return False
class Menus(BaseMenu):
def __init__(self):
super().__init__(main_menu=self.main)
self.mode_handlers = {
"user": self.user,
"repo": self.repo,
"org": self.org,
"search": self.search,
}
def main(self): def main(self):
"""Main menu to select mode.""" """Main menu to select mode."""
set_menu_title(menu_type="home") set_menu_title(menu_type="home")
@@ -207,6 +297,11 @@ class Menus:
value="search", value="search",
description="Search GitHub", description="Search GitHub",
), ),
q.Choice(
title="Clear cache",
value="clear_cache",
description="Clear all in-memory cache from Octosuite",
),
q.Choice( q.Choice(
title="Updates", title="Updates",
value="updates", value="updates",
@@ -239,6 +334,10 @@ class Menus:
elif action == "updates": elif action == "updates":
check_updates() check_updates()
self.main() self.main()
elif action == "clear_cache":
if dialogs.clear_cache():
cache.clear()
self.main()
elif action is None: elif action is None:
sys.exit() sys.exit()
else: else:
@@ -262,9 +361,7 @@ class Menus:
style=CUSTOM_STYLE, style=CUSTOM_STYLE,
) )
clear_screen()
ascii_banner(text=query) ascii_banner(text=query)
option = q.select( option = q.select(
"What would you like to do/search?", "What would you like to do/search?",
choices=[ choices=[
@@ -302,7 +399,7 @@ class Menus:
q.Choice( q.Choice(
title="Go Back", title="Go Back",
value="back", value="back",
description="Go back to ewous menu", description="Go back to previous menu",
shortcut_key="b", shortcut_key="b",
), ),
q.Choice( q.Choice(
@@ -320,9 +417,10 @@ class Menus:
if option is None: if option is None:
self.main() self.main()
return
# Handle navigation # Handle navigation
if self._navigation(option, self.search, query): if self.navigation_handler(option, self.search, query):
return return
# Handle change query # Handle change query
@@ -332,11 +430,9 @@ class Menus:
# Execute search if it's a valid method # Execute search if it's a valid method
if option in self.search_methods: if option in self.search_methods:
with console.status( with Status(
status=f"[dim]Initialising {option} search...[/dim]" status=f"[dim]Initialising {option} search...[/dim]", console=console
) as status: ) as status:
# Get pagination params
status.stop() status.stop()
params = prompts.pagination_params() params = prompts.pagination_params()
status.start() status.start()
@@ -355,15 +451,16 @@ class Menus:
if data: if data:
items = data.get("items") items = data.get("items")
status.stop() status.stop()
self.response_handling( self.response_handler(
data=items if items is not None else data, data=items if items is not None else data,
data_type=option, data_type=option,
source=query, source=query,
) )
# After handling response, show menu again
self.search(query=query) self.search(query=query)
def user(self, username: t.Optional[str] = None): def user(self, username: t.Optional[str] = None, is_validated: bool = False):
"""User menu for querying user data.""" """User menu for querying user data."""
set_menu_title(menu_type="user") set_menu_title(menu_type="user")
clear_screen() clear_screen()
@@ -376,8 +473,15 @@ class Menus:
qmark="@", qmark="@",
) )
clear_screen() user = User(name=username)
ascii_banner(text=username)
if not is_validated:
if not self.target_validator(
identifier=username,
instance=user,
callback=self.user,
):
return
option = q.select( option = q.select(
"What would you like to do/get?", "What would you like to do/get?",
@@ -459,8 +563,10 @@ class Menus:
if option is None: if option is None:
self.main() self.main()
return
# Handle navigation # Handle navigation
if self._navigation(option, self.user, username): if self.navigation_handler(option, self.user, username):
return return
# Handle change username # Handle change username
@@ -468,40 +574,23 @@ class Menus:
self.user() self.user()
return return
# Execute action if it's a valid method # Execute action and handle response
valid_methods = self.paginated_methods | self.non_paginated_methods self.execute_and_handle_response(
if option in valid_methods: instance=user,
with Status( method_name=option,
status=f"[dim]Initialising user {option}...[/dim]", target_type="user",
console=console, source=username,
) as status: )
user = User(name=username)
status.update(f"[dim]Validating user's ({username}) existence...[/dim]") # After handling response, show menu again WITHOUT re-validating
if user.exists(): self.user(username=username, is_validated=True)
console.print(
f"[bold][green]✔[/green] User ({username}) exists on GitHub[/bold]"
)
data = self._execute_selection(
source=username,
instance=user,
method_name=option,
status=status,
)
status.stop() def repo(
self.response_handling(data=data, data_type=option, source=username) self,
else: name: t.Optional[str] = None,
status.stop() owner: t.Optional[str] = None,
console.print( is_validated: bool = False,
f"[bold][yellow]✘[/yellow] User ({username}) doesn't exist on GitHub[/bold]" ):
)
console.input(" Press [bold]ENTER[/bold] to continue ...")
self.user()
self.user(username=username)
def repo(self, name: t.Optional[str] = None, owner: t.Optional[str] = None):
"""Repository menu for querying repo data.""" """Repository menu for querying repo data."""
set_menu_title(menu_type="repo") set_menu_title(menu_type="repo")
clear_screen() clear_screen()
@@ -521,8 +610,17 @@ class Menus:
qmark="@", qmark="@",
) )
clear_screen() repo = Repo(name=name, owner=owner)
ascii_banner(text=f"{owner}/{name}") source = f"{owner}/{name}"
# Only validate if not already validated
if not is_validated:
if not self.target_validator(
identifier=source,
instance=repo,
callback=self.repo,
):
return
option = q.select( option = q.select(
"What would you like to do/get?", "What would you like to do/get?",
@@ -643,9 +741,10 @@ class Menus:
if option is None: if option is None:
self.main() self.main()
return
# Handle navigation # Handle navigation
if self._navigation(option, self.repo, name, owner): if self.navigation_handler(option, self.repo, name, owner):
return return
# Handle change options # Handle change options
@@ -659,40 +758,18 @@ class Menus:
change_handlers[option]() change_handlers[option]()
return return
# Execute action if it's a valid method # Execute action and handle response
valid_methods = self.paginated_methods | self.non_paginated_methods self.execute_and_handle_response(
if option in valid_methods: instance=repo,
source = f"{owner}/{name}" method_name=option,
with Status( target_type="repo",
status=f"[dim]Initialising repository {option}...[/dim]", source=source,
console=console, )
) as status:
repo = Repo(name=name, owner=owner)
status.update( # After handling response, show menu again WITHOUT re-validating
f"[dim]Validating repository's ({source}) existence...[/dim]" self.repo(name=name, owner=owner, is_validated=True)
)
if repo.exists():
console.print(
f"[bold][green]✔[/green] Repository ({source}) exists on GitHub[/bold]"
)
data = self._execute_selection(
source=source, instance=repo, method_name=option, status=status
)
status.stop() def org(self, name: t.Optional[str] = None, is_validated: bool = False):
self.response_handling(data=data, data_type=option, source=source)
else:
status.stop()
console.print(
f"[bold][yellow]✘[/yellow] Repository ({source}) doesn't exist on GitHub[/bold]"
)
console.input(" Press [bold]ENTER[/bold] to continue ...")
self.repo()
self.repo(name=name, owner=owner)
def org(self, name: t.Optional[str] = None):
"""Organisation menu for querying org data.""" """Organisation menu for querying org data."""
set_menu_title(menu_type="org") set_menu_title(menu_type="org")
clear_screen() clear_screen()
@@ -705,8 +782,16 @@ class Menus:
qmark="@", qmark="@",
) )
clear_screen() org = Org(name=name)
ascii_banner(text=name)
# Only validate if not already validated
if not is_validated:
if not self.target_validator(
identifier=name,
instance=org,
callback=self.org,
):
return
option = q.select( option = q.select(
"What would you like to do?", "What would you like to do?",
@@ -759,6 +844,7 @@ class Menus:
shortcut_key="q", shortcut_key="q",
), ),
], ],
pointer=POINTER,
style=CUSTOM_STYLE, style=CUSTOM_STYLE,
instruction=INSTRUCTIONS, instruction=INSTRUCTIONS,
use_shortcuts=True, use_shortcuts=True,
@@ -766,9 +852,10 @@ class Menus:
if option is None: if option is None:
self.main() self.main()
return
# Handle navigation # Handle navigation
if self._navigation(option, self.org, name): if self.navigation_handler(option, self.org, name):
return return
# Handle change org # Handle change org
@@ -776,35 +863,10 @@ class Menus:
self.org() self.org()
return return
# Execute action if it's a valid method # Execute action and handle response
valid_methods = self.paginated_methods | self.non_paginated_methods self.execute_and_handle_response(
if option in valid_methods: instance=org, method_name=option, target_type="org", source=name
with Status( )
status=f"[dim]Initialising organisation {option}...[/dim]",
console=console,
) as status:
org = Org(name=name)
status.update( # After handling response, show menu again WITHOUT re-validating
f"[dim]Validating organisation's ({name}) existence...[/dim]" self.org(name=name, is_validated=True)
)
if org.exists():
console.print(
f"[bold][green]✔[/green] Organisation ({name}) exists on GitHub[/bold]"
)
data = self._execute_selection(
source=name, instance=org, method_name=option, status=status
)
status.stop()
self.response_handling(data=data, data_type=option, source=name)
else:
status.stop()
console.print(
f"[bold][yellow]✘[/yellow] Organisation ({name}) doesn't exist on GitHub[/bold]"
)
console.input(" Press [bold]ENTER[/bold] to continue ...")
self.org()
self.org(name=name)