From ba8ed6bd4ea40e4bf98b160f9a04a76692a5dbbc Mon Sep 17 00:00:00 2001 From: Ritchie Mwewa <74001397+rly0nheart@users.noreply.github.com> Date: Sun, 4 Jan 2026 05:52:01 +0200 Subject: [PATCH] Patches in menus --- pyproject.toml | 2 +- src/octosuite/__init__.py | 2 +- src/octosuite/core/cache.py | 42 +++- src/octosuite/core/github.py | 34 ++- src/octosuite/core/models.py | 343 +++++++++++++++++++++++++++++- src/octosuite/lib.py | 42 +++- src/octosuite/tui/dialogs.py | 27 ++- src/octosuite/tui/menus.py | 397 ++++++++++++++++++++--------------- src/octosuite/tui/prompts.py | 32 ++- 9 files changed, 730 insertions(+), 191 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 5bde0b0..7be7984 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "octosuite" -version = "4.0.0beta1" +version = "4.0.0beta2" description = "TUI-based toolkit for GitHub data analysis." readme = "README.md" license = "MIT" diff --git a/src/octosuite/__init__.py b/src/octosuite/__init__.py index 9bbbd90..0fec0ed 100644 --- a/src/octosuite/__init__.py +++ b/src/octosuite/__init__.py @@ -1,2 +1,2 @@ __pkg__ = "octosuite" -__version__ = "4.0.0beta1" +__version__ = "4.0.0beta2" diff --git a/src/octosuite/core/cache.py b/src/octosuite/core/cache.py index 12960ec..345cdb8 100644 --- a/src/octosuite/core/cache.py +++ b/src/octosuite/core/cache.py @@ -8,30 +8,59 @@ class ResponseCache: """Simple in-memory cache for API responses.""" def __init__(self): + """Initialise the ResponseCache with an empty cache dictionary.""" + self._cache = {} @staticmethod def _generate_key(url: str, params: dict = None) -> str: - """Generate a unique cache key from URL and params.""" + """ + Generate a unique cache key from URL and parameters. + + :param url: The URL to generate a key for. + :param params: Optional dictionary of parameters to include in the key. + :return: MD5 hash string representing the unique cache key. + """ + 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.""" + """ + Retrieve a cached response if it exists. + + :param url: The URL to retrieve cached data for. + :param params: Optional dictionary of parameters used in the original request. + :return: Cached response data if found, None otherwise. + """ + key = self._generate_key(url, params) return self._cache.get(key) def set(self, url: str, data, params: dict = None): - """Cache a response.""" + """ + Store a response in the cache. + + :param url: The URL to cache data for. + :param data: The response data to cache. + :param params: Optional dictionary of parameters used in the request. + """ + key = self._generate_key(url, params) self._cache[key] = data def clear(self): - """Clear all cached responses.""" + """Clear all cached responses from memory.""" self._cache.clear() def remove(self, url: str, params: dict = None): - """Remove a specific cached response.""" + """ + Remove a specific cached response. + + :param url: The URL to remove cached data for. + :param params: Optional dictionary of parameters used in the original request. + """ + key = self._generate_key(url, params) self._cache.pop(key, None) @@ -40,5 +69,6 @@ cache = ResponseCache() def clear_cache(): - """Clear all cached API responses.""" + """Clear all cached API responses from the global cache instance.""" + cache.clear() diff --git a/src/octosuite/core/github.py b/src/octosuite/core/github.py index 83b8bb4..d3e993c 100644 --- a/src/octosuite/core/github.py +++ b/src/octosuite/core/github.py @@ -14,6 +14,8 @@ __all__ = ["BASE_URL", "GitHub"] class GitHub: + """Handles GitHub API requests with caching and response sanitisation.""" + def __init__( self, user_agent: str = ( @@ -22,6 +24,12 @@ class GitHub: f"https://github.com/bellingcat/octosuite) requests/{requests.__version__}" ), ): + """ + Initialise the GitHub API client. + + :param user_agent: Custom User-Agent string for API requests. + """ + self.user_agent = user_agent self.cache = cache @@ -32,6 +40,16 @@ class GitHub: return_response: bool = False, use_cache: bool = True, ) -> t.Union[dict, list, Response]: + """ + Make a GET request to the GitHub API. + + :param url: The API endpoint URL. + :param params: Optional query parameters for the request. + :param return_response: If True, return the raw Response object instead of JSON data. + :param use_cache: If True, use cached responses when available. + :return: Dictionary, list, or Response object depending on the request and parameters. + """ + if use_cache and not return_response: cached = self.cache.get(url, params) if cached is not None: @@ -58,7 +76,14 @@ class GitHub: def is_valid_entity( self, _type: t.Literal["user", "org", "repo"], **kwargs ) -> bool: - """Validate if a GitHub entity exists.""" + """ + Validate whether a GitHub entity exists. + + :param _type: Type of entity to validate ("user", "org", or "repo"). + :param kwargs: Entity identifiers (username for user/org, repo_owner and repo_name for repo). + :return: True if the entity exists, False otherwise. + """ + try: type_map = { "user": f"https://api.github.com/users/{kwargs.get('username')}", @@ -86,6 +111,13 @@ class GitHub: return False def sanitise_response(self, response: t.Union[dict, list]) -> t.Union[dict, list]: + """ + Remove API URLs and null values from response data recursively. + + :param response: The response data to sanitise (dict or list). + :return: Sanitised response with API URLs and null values removed. + """ + pattern = re.compile(r"https://api\.github\.com") if isinstance(response, list): diff --git a/src/octosuite/core/models.py b/src/octosuite/core/models.py index d7aa6bf..034948d 100644 --- a/src/octosuite/core/models.py +++ b/src/octosuite/core/models.py @@ -8,14 +8,25 @@ __all__ = ["User", "Org", "Repo", "Search"] class GitHubEntity: - """Base class for GitHub entities with common functionality.""" + """ + Base class for GitHub entities with common functionality.""" def __init__(self, source: str): + """ + Initialise a GitHub entity. + + :param source: The source identifier for the entity. + """ + self.endpoint = None self.source = source def exists(self) -> tuple[bool, dict]: - """Check if the entity exists on GitHub.""" + """ + Check if the entity exists on GitHub. + + :return: Tuple of (exists, response_data) where exists is True if the entity is found. + """ # Check cache first cached = github.cache.get(self.endpoint) if cached is not None: @@ -38,58 +49,150 @@ class GitHubEntity: class User(GitHubEntity): + """Represents a GitHub user with methods to query user data.""" + def __init__(self, name: str): + """ + Initialise a User instance. + + :param name: The GitHub username. + """ + super().__init__(source=name) self.name = name self.endpoint = f"{BASE_URL}/users/{name}" def profile(self) -> dict: + """ + Retrieve the user's profile information. + + :return: Dictionary containing user profile data. + """ + profile = github.get(url=self.endpoint) return profile def repos(self, page: int, per_page: int) -> list: + """ + Retrieve the user's public repositories. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of repository dictionaries. + """ + params = {"page": page, "per_page": per_page} repos = github.get(url=f"{self.endpoint}/repos", params=params) return repos def subscriptions(self, page: int, per_page: int) -> list: + """ + Retrieve repositories the user is subscribed to. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of repository subscription dictionaries. + """ + params = {"page": page, "per_page": per_page} subscriptions = github.get(url=f"{self.endpoint}/subscriptions", params=params) return subscriptions def starred(self, page: int, per_page: int) -> list: + """ + Retrieve repositories the user has starred. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of starred repository dictionaries. + """ + params = {"page": page, "per_page": per_page} starred = github.get(url=f"{self.endpoint}/starred", params=params) return starred def followers(self, page: int, per_page: int) -> list: + """ + Retrieve the user's followers. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of follower user dictionaries. + """ + params = {"page": page, "per_page": per_page} users = github.get(url=f"{self.endpoint}/followers", params=params) return users def following(self, page: int, per_page: int) -> list: + """ + Retrieve users that this user is following. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of followed user dictionaries. + """ + params = {"page": page, "per_page": per_page} users = github.get(url=f"{self.endpoint}/following", params=params) return users - def follows(self, user: str) -> bool: ... + def follows(self, user: str) -> bool: + """Check if this user follows another user. + + :param user: The username to check. + :return: True if this user follows the specified user. + """ + ... def orgs(self, page: int, per_page: int) -> list: + """ + Retrieve organisations the user belongs to. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of organisation dictionaries. + """ + params = {"page": page, "per_page": per_page} orgs = github.get(url=f"{self.endpoint}/orgs", params=params) return orgs def gists(self, page: int, per_page: int) -> list: + """ + Retrieve the user's gists. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of gist dictionaries. + """ + params = {"page": page, "per_page": per_page} gists = github.get(url=f"{self.endpoint}/gists", params=params) return gists def events(self, page: int, per_page: int) -> list: + """ + Retrieve the user's public events. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of event dictionaries. + """ + params = {"page": page, "per_page": per_page} events = github.get(url=f"{self.endpoint}/events", params=params) return events def received_events(self, page: int, per_page: int) -> list: + """ + Retrieve events received by the user. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of received event dictionaries. + """ + params = {"page": page, "per_page": per_page} received_events = github.get( url=f"{self.endpoint}/received_events", params=params @@ -98,133 +201,337 @@ class User(GitHubEntity): class Org(GitHubEntity): + """Represents a GitHub organisation with methods to query organisation data.""" + def __init__(self, name: str): + """ + Initialise an Org instance. + + :param name: The GitHub organisation name. + """ + super().__init__(source=name) self.name = name self.endpoint = f"{BASE_URL}/orgs/{name}" def profile(self) -> dict: + """ + Retrieve the organisation's profile information. + + :return: Dictionary containing organisation profile data. + """ + profile = github.get(url=self.endpoint) return profile def repos(self, page: int, per_page: int) -> list: + """ + Retrieve the organisation's public repositories. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of repository dictionaries. + """ + params = {"page": page, "per_page": per_page} repos = github.get(url=f"{self.endpoint}/repos", params=params) return repos def events(self, page: int, per_page: int) -> list: + """ + Retrieve the organisation's public events. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of event dictionaries. + """ + params = {"page": page, "per_page": per_page} events = github.get(url=f"{self.endpoint}/events", params=params) return events def hooks(self, page: int, per_page: int) -> list: + """ + Retrieve the organisation's webhooks. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of webhook dictionaries. + """ + params = {"page": page, "per_page": per_page} hooks = github.get(url=f"{self.endpoint}/hooks", params=params) return hooks def issues(self, page: int, per_page: int) -> list: + """ + Retrieve the organisation's issues. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of issue dictionaries. + """ + params = {"page": page, "per_page": per_page} issues = github.get(url=f"{self.endpoint}/issues", params=params) return issues def members(self, page: int, per_page: int) -> list: + """ + Retrieve the organisation's public members. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of member user dictionaries. + """ + params = {"page": page, "per_page": per_page} members = github.get(url=f"{self.endpoint}/members", params=params) return members class Repo(GitHubEntity): + """Represents a GitHub repository with methods to query repository data.""" + def __init__(self, name: str, owner: str): + """ + Initialise a Repo instance. + + :param name: The repository name. + :param owner: The repository owner's username. + """ + super().__init__(source=f"{owner}/{name}") self.name = name self.owner = owner self.endpoint = f"{BASE_URL}/repos/{owner}/{name}" def profile(self) -> dict: + """ + Retrieve the repository's information. + + :return: Dictionary containing repository data. + """ + profile = github.get(url=self.endpoint) return profile def forks(self, page: int, per_page: int) -> list: + """ + Retrieve the repository's forks. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of fork repository dictionaries. + """ + params = {"page": page, "per_page": per_page} forks = github.get(url=f"{self.endpoint}/forks", params=params) return forks def issue_events(self, page: int, per_page: int) -> list: + """ + Retrieve the repository's issue events. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of issue event dictionaries. + """ + params = {"page": page, "per_page": per_page} issue_events = github.get(url=f"{self.endpoint}/issue_events", params=params) return issue_events def events(self, page: int, per_page: int) -> list: + """ + Retrieve the repository's events. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of event dictionaries. + """ + params = {"page": page, "per_page": per_page} events = github.get(url=f"{self.endpoint}/events", params=params) return events def assignees(self, page: int, per_page: int) -> list: + """ + Retrieve the repository's available assignees. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of assignee user dictionaries. + """ + params = {"page": page, "per_page": per_page} assignees = github.get(url=f"{self.endpoint}/assignees", params=params) return assignees def branches(self, page: int, per_page: int) -> list: + """ + Retrieve the repository's branches. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of branch dictionaries. + """ + params = {"page": page, "per_page": per_page} branches = github.get(url=f"{self.endpoint}/branches", params=params) return branches def tags(self, page: int, per_page: int) -> list: + """ + Retrieve the repository's tags. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of tag dictionaries. + """ + params = {"page": page, "per_page": per_page} tags = github.get(url=f"{self.endpoint}/tags", params=params) return tags def languages(self) -> dict: + """ + Retrieve the programming languages used in the repository. + + :return: Dictionary mapping language names to bytes of code. + """ + languages = github.get(url=f"{self.endpoint}/languages") return languages def stargazers(self, page: int, per_page: int) -> list: + """ + Retrieve users who have starred the repository. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of stargazer user dictionaries. + """ + params = {"page": page, "per_page": per_page} stargazers = github.get(url=f"{self.endpoint}/stargazers", params=params) return stargazers def subscribers(self, page: int, per_page: int) -> list: + """ + Retrieve users subscribed to the repository. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of subscriber user dictionaries. + """ + params = {"page": page, "per_page": per_page} subscribers = github.get(url=f"{self.endpoint}/subscribers", params=params) return subscribers def commits(self, page: int, per_page: int) -> list: + """ + Retrieve the repository's commits. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of commit dictionaries. + """ + params = {"page": page, "per_page": per_page} commits = github.get(url=f"{self.endpoint}/commits", params=params) return commits def comments(self, page: int, per_page: int) -> list: + """ + Retrieve the repository's commit comments. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of comment dictionaries. + """ + params = {"page": page, "per_page": per_page} comments = github.get(url=f"{self.endpoint}/comments", params=params) return comments def contents(self, path: str) -> list: + """ + Retrieve the contents of a file or directory in the repository. + + :param path: Path to the file or directory. + :return: List of content item dictionaries. + """ + contents = github.get(url=f"{self.endpoint}/contents/{path}") return contents def issues(self, page: int, per_page: int) -> list: + """ + Retrieve the repository's issues. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of issue dictionaries. + """ + params = {"page": page, "per_page": per_page} issues = github.get(url=f"{self.endpoint}/issues", params=params) return issues def releases(self, page: int, per_page: int) -> list: + """ + Retrieve the repository's releases. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of release dictionaries. + """ + params = {"page": page, "per_page": per_page} releases = github.get(url=f"{self.endpoint}/releases", params=params) return releases def deployments(self, page: int, per_page: int) -> list: + """ + Retrieve the repository's deployments. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of deployment dictionaries. + """ + params = {"page": page, "per_page": per_page} deployments = github.get(url=f"{self.endpoint}/deployments", params=params) return deployments def labels(self, page: int, per_page: int) -> list: + """ + Retrieve the repository's labels. + + :param page: Page number for pagination. + :param per_page: Number of results per page. + :return: List of label dictionaries. + """ + params = {"page": page, "per_page": per_page} labels = github.get(url=f"{self.endpoint}/labels", params=params) return labels class Search: + """Provides methods to search GitHub for various entity types.""" + def __init__(self, query: str, page: int, per_page: int): + """ + Initialise a Search instance. + + :param query: The search query string. + :param page: Page number for pagination. + :param per_page: Number of results per page. + """ + self.query = query self.page = page self.per_page = per_page @@ -232,21 +539,51 @@ class Search: self.params = {"q": query, "page": page, "per_page": per_page} def repos(self) -> list: + """ + Search for repositories matching the query. + + :return: List of repository search result dictionaries. + """ + repos = github.get(url=f"{self.endpoint}/repositories", params=self.params) return repos def users(self) -> list: + """ + Search for users matching the query. + + :return: List of user search result dictionaries. + """ + users = github.get(url=f"{self.endpoint}/users", params=self.params) return users def commits(self) -> list: + """ + Search for commits matching the query. + + :return: List of commit search result dictionaries. + """ + commits = github.get(url=f"{self.endpoint}/commits", params=self.params) return commits def issues(self) -> list: + """ + Search for issues and pull requests matching the query. + + :return: List of issue search result dictionaries. + """ + issues = github.get(url=f"{self.endpoint}/issues", params=self.params) return issues def topics(self) -> list: + """ + Search for topics matching the query. + + :return: List of topic search result dictionaries. + """ + topics = github.get(url=f"{self.endpoint}/topics", params=self.params) return topics diff --git a/src/octosuite/lib.py b/src/octosuite/lib.py index 519024e..b47c2a4 100644 --- a/src/octosuite/lib.py +++ b/src/octosuite/lib.py @@ -31,6 +31,14 @@ console = Console(log_time=False) def preview_response(data: t.Union[dict, list], source: str, _type: str): + """ + Display a preview of response data in a tree structure. + + :param data: The data to preview (dict or list). + :param source: The source identifier of the data. + :param _type: The type of data being previewed. + """ + if isinstance(data, dict): tree = Tree( label=f"\n[bold]{data.get('name') or data.get('login') or data.get('id') or 'Data'}[/bold]", @@ -44,7 +52,7 @@ def preview_response(data: t.Union[dict, list], source: str, _type: str): elif isinstance(data, list): preview_data = data[:5] tree = Tree( - label=f"\nPreview: First {len(preview_data)} of {len(data)} {_type} from {source}", + label=f"\n[bold]First {len(preview_data)} of {len(data)} {_type} for '{source}'[/bold]", guide_style="#444444", highlight=True, ) @@ -74,7 +82,17 @@ def export_response( file_formats: list, output_dir: str = "../exports", ): - """Export data to selected formats using built-in libraries.""" + """ + Export response data to one or more file formats. + + :param data: The data to export (dict or list). + :param data_type: The type of data being exported. + :param source: The source identifier of the data. + :param file_formats: List of file formats to export to (json, csv, html). + :param output_dir: Directory path where exported files will be saved. + :return: List of exported file paths. + """ + # Create output directory if it doesn't exist output_dir = Path(output_dir) output_dir.mkdir(exist_ok=True) @@ -165,6 +183,14 @@ def export_response( def fill_tree(tree: Tree, data: t.Union[dict, list]) -> Tree: + """ + Recursively populate a Rich Tree with data. + + :param tree: The Tree object to populate. + :param data: The data to add to the tree (dict or list). + :return: The populated Tree object. + """ + if isinstance(data, dict): for key, value in data.items(): if isinstance(value, dict) or isinstance(value, list): @@ -187,6 +213,8 @@ def fill_tree(tree: Tree, data: t.Union[dict, list]) -> Tree: def check_updates(): + """Check for available package updates and display the result.""" + with console.status("[dim]Checking for updates...[/dim]") as status: checker = UpdateChecker() result = checker.check(__pkg__, __version__) @@ -207,6 +235,10 @@ def clear_screen(): def ascii_banner(text: str): + """Display a colourful ASCII art banner with gradient styling. + + :param text: The text to convert to ASCII art. + """ clear_screen() ascii_text = pyfiglet.figlet_format(text=text, font="chunky") @@ -224,6 +256,12 @@ def ascii_banner(text: str): def set_menu_title(menu_type: t.Literal["home", "user", "org", "repo", "search"]): + """ + Set the terminal window title based on the current menu. + + :param menu_type: The type of menu being displayed. + """ + title: str = __pkg__.title() title += f" | {menu_type.title()}" console.set_window_title(title=title) diff --git a/src/octosuite/tui/dialogs.py b/src/octosuite/tui/dialogs.py index 742bb74..7cab040 100644 --- a/src/octosuite/tui/dialogs.py +++ b/src/octosuite/tui/dialogs.py @@ -26,10 +26,22 @@ __all__ = ["Dialogs"] class Dialogs: - def __init__(self): ... + """Provides interactive dialogue boxes for user confirmations and information display.""" + + def __init__(self): + """Initialise the Dialogs class.""" + ... @staticmethod def _boolean(title: str, text: str) -> bool: + """ + Display a Yes/No dialogue and return the user's choice. + + :param title: The title of the dialogue box. + :param text: The message text to display. + :return: True if user selects Yes, False if No or cancelled. + """ + try: result = button_dialog( title=title, @@ -41,11 +53,23 @@ class Dialogs: return True def quit(self) -> bool: + """ + Display a confirmation dialogue for quitting the application. + + :return: True if user confirms quit, False otherwise. + """ + return self._boolean( title="Quit", text="This will close the session, continue?" ) def clear_cache(self) -> bool: + """ + Display a confirmation dialogue for clearing the cache. + + :return: True if user confirms cache clearing, False otherwise. + """ + return self._boolean( title="Clear Cache", text="This will clear all octosuite caches, continue?", @@ -53,4 +77,5 @@ class Dialogs: @staticmethod def license(): + """Display the MIT license notice in a dialogue box.""" message_dialog(title="MIT License", text=LICENSE_NOTICE).run() diff --git a/src/octosuite/tui/menus.py b/src/octosuite/tui/menus.py index b231fb5..34f9950 100644 --- a/src/octosuite/tui/menus.py +++ b/src/octosuite/tui/menus.py @@ -21,7 +21,6 @@ CUSTOM_STYLE = Style( INSTRUCTIONS = "↑↓ [move] • ⮠ [select]" EXPORT_INSTRUCTIONS = "↑↓ [move] • ⮠ [confirm] • spacebar [check]" -POINTER: str = "🖝 " dialogs = Dialogs() prompts = Prompts() @@ -30,9 +29,15 @@ __all__ = ["Menus"] class BaseMenu: - """Base class with common menu functionality.""" + """Base class providing common menu functionality and response handling for GitHub data queries.""" def __init__(self, main_menu: t.Callable): + """ + Initialise the BaseMenu with pagination and search method configurations. + + :param main_menu: Callable reference to the main menu function. + """ + # Define which methods require pagination self.paginated_methods = { "repos", @@ -76,17 +81,16 @@ class BaseMenu: 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 """ + Validate whether a GitHub entity exists. + + :param identifier: Name or identifier of the entity to validate. + :param instance: The instance with an exists() method (User, Repo, or Org). + :param callback: Function to call if validation fails. + :param callback_args: Arguments to pass to the callback function. + :return: True if the entity exists, False otherwise. + """ + if isinstance(instance, User): target_type = "user" elif isinstance(instance, Repo): @@ -95,18 +99,12 @@ class BaseMenu: target_type = "org" with Status( - f"[dim]Validating {target_type}'s ({identifier}) existence...[/dim]", + f"[dim]Validating {target_type} ({identifier})[/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: + if not exists: status.stop() if response and "message" in response: console.print( @@ -116,6 +114,8 @@ class BaseMenu: callback(*callback_args) return False + return True + def execute_and_handle_response( self, instance: t.Union[User, Repo, Org, Search], @@ -123,18 +123,19 @@ class BaseMenu: 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 """ + Execute a method on an instance and handle the resulting data. + + :param instance: The instance to execute the method on. + :param method_name: Name of the method to execute. + :param target_type: Type of entity ("user", "repo", or "org"). + :param 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]", + status=f"[dim]Initialising {target_type} {method_name}[/dim]...", console=console, ) as status: data = self.execute_selection( @@ -148,7 +149,14 @@ class BaseMenu: 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 with pagination if required. + + :param status: Rich Status object for displaying progress. + :param kwargs: Must include 'instance', 'method_name', and 'source'. + :return: The data returned by the executed method. + """ + instance = kwargs.get("instance") method_name = kwargs.get("method_name") source = kwargs.get("source") @@ -160,88 +168,111 @@ class BaseMenu: prompts.pagination_params() if method_name in self.paginated_methods else {} ) status.start() - status.update(f"[dim]Getting {method_name} from {source}...[/dim]") + status.update(f"[dim]Getting {method_name} from {source}[/dim]...") return method(**params) def response_handler(self, data: t.Union[dict, list], data_type: str, source: str): - """Export data to file in user-selected format(s).""" - preview_response(data=data, source=source, _type=data_type) + """ + Handle retrieved data by previewing and offering export options. + + :param data: The data to handle (dict or list). + :param data_type: Type of data retrieved. + :param source: Source identifier of the data. + """ try: - export_choice = q.select( - "What would you like to do?", - choices=[ - q.Choice( - title="Export", - value="export", - description="Export the data", - shortcut_key="e", - ), - q.Choice( - title="Skip", - value="skip", - description="Do nothing, and go back to previous menu", - shortcut_key="x", - ), - q.Choice( - title="Quit", - value="quit", - description="Close this session", - ), - ], - pointer=POINTER, - style=CUSTOM_STYLE, - instruction=INSTRUCTIONS, - ).ask() + if not data: + console.print( + f"[bold][yellow]✘[/yellow] No data found for '{source}'[/bold]" + ) + else: + preview_response(data=data, source=source, _type=data_type) + export_choice = q.select( + "What would you like to do?", + choices=[ + q.Choice( + title="Export", + value="export", + description="Export the data", + shortcut_key="e", + ), + q.Choice( + title="Skip", + value="skip", + description="Do nothing, and go back to previous menu", + shortcut_key="x", + ), + q.Choice( + title="Quit", + value="quit", + description="Close this session", + ), + ], + style=CUSTOM_STYLE, + instruction=INSTRUCTIONS, + ).ask() - if export_choice == "skip": - return - - if export_choice == "quit": - if dialogs.quit(): - sys.exit() - else: - self.response_handler( - data=data, - data_type=data_type, - source=source, - ) + if export_choice == "skip": return - file_formats = q.checkbox( - "Select export format(s)", - choices=[ - q.Choice( - title="JSON", - value="json", - description="Export as JSON file", - ), - q.Choice( - title="CSV", - value="csv", - description="Export as CSV file", - ), - q.Choice( - title="HTML", - value="html", - description="Export as HTML table", - ), - ], - pointer=POINTER, - style=CUSTOM_STYLE, - instruction=EXPORT_INSTRUCTIONS, - validate=lambda x: len(x) > 0 or "Please select at least one format", - ).ask() + if export_choice == "quit": + if dialogs.quit(): + sys.exit() + else: + clear_screen() + self.response_handler( + data=data, + data_type=data_type, + source=source, + ) + return - export_response( - data=data, data_type=data_type, source=source, file_formats=file_formats - ) + file_formats = q.checkbox( + "Select export format(s)", + choices=[ + q.Choice( + title="JSON", + value="json", + description="Export as JSON file", + ), + q.Choice( + title="CSV", + value="csv", + description="Export as CSV file", + ), + q.Choice( + title="HTML", + value="html", + description="Export as HTML table", + ), + ], + style=CUSTOM_STYLE, + instruction=EXPORT_INSTRUCTIONS, + validate=lambda x: len(x) > 0 + or "Please select at least one format", + ).ask() + export_response( + data=data, + data_type=data_type, + source=source, + file_formats=file_formats, + ) except KeyboardInterrupt: - print("\nExport cancelled") + console.print("\nExport cancelled") + finally: + console.input(" Press [bold]ENTER[/bold] to continue ...") def navigation_handler(self, option: str, callback: t.Callable, *callback_args): - """Handle navigation options (back, quit, change settings).""" + """ + Handle common navigation options across menus. + + :param option: The selected navigation option. + :param callback: Function to call for certain navigation actions. + :param callback_args: Arguments to pass to the callback function. + :return: True if navigation was handled, False otherwise. + """ + navigation_handlers = { "back": lambda: self.main_menu(), "quit": lambda: ( @@ -257,7 +288,11 @@ class BaseMenu: class Menus(BaseMenu): + """Main menu system providing interactive interfaces for GitHub data queries.""" + def __init__(self): + """Initialise the Menus class with mode handlers for different query types.""" + super().__init__(main_menu=self.main) self.mode_handlers = { @@ -268,7 +303,8 @@ class Menus(BaseMenu): } def main(self): - """Main menu to select mode.""" + """Display the main menu for selecting query mode.""" + set_menu_title(menu_type="home") clear_screen() try: @@ -318,7 +354,6 @@ class Menus(BaseMenu): description="Close this session", ), ], - pointer=POINTER, style=CUSTOM_STYLE, instruction=INSTRUCTIONS, ).ask() @@ -350,7 +385,12 @@ class Menus(BaseMenu): sys.exit() def search(self, query: t.Optional[str] = None): - """Search menu for querying GitHub.""" + """ + Display the search menu for querying GitHub content. + + :param query: Optional search query string. If None, prompts user for input. + """ + set_menu_title(menu_type="search") clear_screen() if query is None: @@ -409,7 +449,6 @@ class Menus(BaseMenu): shortcut_key="q", ), ], - pointer=POINTER, style=CUSTOM_STYLE, instruction=INSTRUCTIONS, use_shortcuts=True, @@ -431,7 +470,7 @@ class Menus(BaseMenu): # Execute search if it's a valid method if option in self.search_methods: with Status( - status=f"[dim]Initialising {option} search...[/dim]", console=console + status=f"[dim]Initialising {option} search[/dim]...", console=console ) as status: status.stop() params = prompts.pagination_params() @@ -445,7 +484,7 @@ class Menus(BaseMenu): ) method = getattr(search, option) - status.update(f"[dim]Searching {option} for {query}...[/dim]") + status.update(f"[dim]Searching {option} for {query}[/dim]...") data = method() if data: @@ -461,7 +500,13 @@ class Menus(BaseMenu): self.search(query=query) def user(self, username: t.Optional[str] = None, is_validated: bool = False): - """User menu for querying user data.""" + """ + Display the user menu for querying GitHub user data. + + :param username: Optional GitHub username. If None, prompts user for input. + :param is_validated: Whether the username has already been validated. + """ + set_menu_title(menu_type="user") clear_screen() if username is None: @@ -475,14 +520,14 @@ class Menus(BaseMenu): user = User(name=username) - if not is_validated: - if not self.target_validator( - identifier=username, - instance=user, - callback=self.user, - ): - return + if not self.target_validator( + identifier=username, + instance=user, + callback=self.user, + ): + return + ascii_banner(text=username) option = q.select( "What would you like to do/get?", choices=[ @@ -555,7 +600,6 @@ class Menus(BaseMenu): shortcut_key="q", ), ], - pointer=POINTER, style=CUSTOM_STYLE, instruction=INSTRUCTIONS, use_shortcuts=True, @@ -566,32 +610,37 @@ class Menus(BaseMenu): return # Handle navigation - if self.navigation_handler(option, self.user, username): + elif self.navigation_handler(option, self.user, username): return # Handle change username - if option == "change_username": + elif option == "change_username": self.user() return + else: + # Execute action and handle response + self.execute_and_handle_response( + instance=user, + method_name=option, + target_type="user", + source=username, + ) - # Execute action and handle response - self.execute_and_handle_response( - instance=user, - method_name=option, - target_type="user", - source=username, - ) - - # After handling response, show menu again WITHOUT re-validating - self.user(username=username, is_validated=True) + # After handling response, show menu again WITHOUT re-validating + self.user(username=username, is_validated=True) def repo( self, name: t.Optional[str] = None, owner: t.Optional[str] = None, - is_validated: bool = False, ): - """Repository menu for querying repo data.""" + """ + Display the repository menu for querying GitHub repository data. + + :param name: Optional repository name. If None, prompts user for input. + :param owner: Optional repository owner. If None, prompts user for input. + """ + set_menu_title(menu_type="repo") clear_screen() if name is None or owner is None: @@ -614,14 +663,14 @@ class Menus(BaseMenu): 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 + if not self.target_validator( + identifier=source, + instance=repo, + callback=self.repo, + ): + return + ascii_banner(text=source) option = q.select( "What would you like to do/get?", choices=[ @@ -733,20 +782,11 @@ class Menus(BaseMenu): shortcut_key="q", ), ], - pointer=POINTER, style=CUSTOM_STYLE, instruction=INSTRUCTIONS, use_shortcuts=True, ).ask() - if option is None: - self.main() - return - - # Handle navigation - if self.navigation_handler(option, self.repo, name, owner): - return - # Handle change options change_handlers = { "change_repo_name": lambda: self.repo(owner=owner), @@ -754,23 +794,36 @@ class Menus(BaseMenu): "change_both": lambda: self.repo(), } - if option in change_handlers: - change_handlers[option]() + if option is None: + self.main() return - # Execute action and handle response - self.execute_and_handle_response( - instance=repo, - method_name=option, - target_type="repo", - source=source, - ) + # Handle navigation + elif self.navigation_handler(option, self.repo, name, owner): + return - # After handling response, show menu again WITHOUT re-validating - self.repo(name=name, owner=owner, is_validated=True) + elif option in change_handlers: + change_handlers[option]() + return + else: + # Execute action and handle response + self.execute_and_handle_response( + instance=repo, + method_name=option, + target_type="repo", + source=source, + ) + + # After handling response, show menu again WITHOUT re-validating + self.repo(name=name, owner=owner) + + def org(self, name: t.Optional[str] = None): + """ + Display the organisation menu for querying GitHub organisation data. + + :param name: Optional organisation name. If None, prompts user for input. + """ - def org(self, name: t.Optional[str] = None, is_validated: bool = False): - """Organisation menu for querying org data.""" set_menu_title(menu_type="org") clear_screen() if name is None: @@ -784,14 +837,12 @@ class Menus(BaseMenu): org = Org(name=name) - # Only validate if not already validated - if not is_validated: - if not self.target_validator( - identifier=name, - instance=org, - callback=self.org, - ): - return + if not self.target_validator( + identifier=name, + instance=org, + callback=self.org, + ): + return option = q.select( "What would you like to do?", @@ -844,7 +895,6 @@ class Menus(BaseMenu): shortcut_key="q", ), ], - pointer=POINTER, style=CUSTOM_STYLE, instruction=INSTRUCTIONS, use_shortcuts=True, @@ -855,18 +905,19 @@ class Menus(BaseMenu): return # Handle navigation - if self.navigation_handler(option, self.org, name): + elif self.navigation_handler(option, self.org, name): return # Handle change org - if option == "change_org": + elif option == "change_org": self.org() return - # Execute action and handle response - self.execute_and_handle_response( - instance=org, method_name=option, target_type="org", source=name - ) + else: + # Execute action and handle response + self.execute_and_handle_response( + instance=org, method_name=option, target_type="org", source=name + ) - # After handling response, show menu again WITHOUT re-validating - self.org(name=name, is_validated=True) + # After handling response, show menu again WITHOUT re-validating + self.org(name=name) diff --git a/src/octosuite/tui/prompts.py b/src/octosuite/tui/prompts.py index e255cf2..889c0b7 100644 --- a/src/octosuite/tui/prompts.py +++ b/src/octosuite/tui/prompts.py @@ -7,7 +7,10 @@ __all__ = ["Prompts"] class Prompts: + """Provides interactive prompts for user input collection.""" + def __init__(self): + """Initialise the Prompts class.""" pass @staticmethod @@ -17,17 +20,40 @@ class Prompts: style: t.Optional[Style] = None, qmark: t.Optional[str] = "?", ) -> str: - return q.text( + """ + Display a text input prompt and validate non-empty input. + + :param message: The prompt message to display. + :param instruction: Optional instruction text to guide the user. + :param style: Optional questionary Style object for custom styling. + :param qmark: Optional custom question mark character or string. + :return: The user's input string. + :raises KeyboardInterrupt: If the user cancels the prompt with CTRL+C. + """ + + result = q.text( message=message, instruction=instruction, style=style, qmark=qmark, - validate=lambda text: len(text.strip()) > 0 or "Input cannot be empty", + validate=lambda text: len(text.strip()) > 0 + or None + or "Input cannot be empty", ).ask() + if result is None: + raise KeyboardInterrupt + + return result + @staticmethod def pagination_params() -> dict: - """Prompt user for pagination parameters.""" + """ + Prompt the user for pagination parameters. + + :return: Dictionary containing 'page' and 'per_page' integer values. + """ + try: page = q.text(message="Page", default="1", qmark="n").ask() per_page = q.text(