mirror of
https://github.com/bellingcat/octosuite.git
synced 2026-06-12 21:38:34 +03:00
Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9b0f490089 | ||
|
|
d9e2dae71d | ||
|
|
0c06b95d18 | ||
|
|
ba8ed6bd4e |
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "octosuite"
|
name = "octosuite"
|
||||||
version = "4.0.0beta1"
|
version = "4.0.0beta2"
|
||||||
description = "TUI-based toolkit for GitHub data analysis."
|
description = "TUI-based toolkit for GitHub data analysis."
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
license = "MIT"
|
license = "MIT"
|
||||||
|
|||||||
@@ -1,2 +1,2 @@
|
|||||||
__pkg__ = "octosuite"
|
__pkg__ = "octosuite"
|
||||||
__version__ = "4.0.0beta1"
|
__version__ = "4.0.0beta2"
|
||||||
|
|||||||
@@ -8,30 +8,59 @@ class ResponseCache:
|
|||||||
"""Simple in-memory cache for API responses."""
|
"""Simple in-memory cache for API responses."""
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
|
"""Initialise the ResponseCache with an empty cache dictionary."""
|
||||||
|
|
||||||
self._cache = {}
|
self._cache = {}
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _generate_key(url: str, params: dict = None) -> str:
|
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)}"
|
cache_data = f"{url}:{json.dumps(params or {}, sort_keys=True)}"
|
||||||
return hashlib.md5(cache_data.encode()).hexdigest()
|
return hashlib.md5(cache_data.encode()).hexdigest()
|
||||||
|
|
||||||
def get(self, url: str, params: dict = None):
|
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)
|
key = self._generate_key(url, params)
|
||||||
return self._cache.get(key)
|
return self._cache.get(key)
|
||||||
|
|
||||||
def set(self, url: str, data, params: dict = None):
|
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)
|
key = self._generate_key(url, params)
|
||||||
self._cache[key] = data
|
self._cache[key] = data
|
||||||
|
|
||||||
def clear(self):
|
def clear(self):
|
||||||
"""Clear all cached responses."""
|
"""Clear all cached responses from memory."""
|
||||||
self._cache.clear()
|
self._cache.clear()
|
||||||
|
|
||||||
def remove(self, url: str, params: dict = None):
|
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)
|
key = self._generate_key(url, params)
|
||||||
self._cache.pop(key, None)
|
self._cache.pop(key, None)
|
||||||
|
|
||||||
@@ -40,5 +69,6 @@ cache = ResponseCache()
|
|||||||
|
|
||||||
|
|
||||||
def clear_cache():
|
def clear_cache():
|
||||||
"""Clear all cached API responses."""
|
"""Clear all cached API responses from the global cache instance."""
|
||||||
|
|
||||||
cache.clear()
|
cache.clear()
|
||||||
|
|||||||
@@ -14,6 +14,8 @@ __all__ = ["BASE_URL", "GitHub"]
|
|||||||
|
|
||||||
|
|
||||||
class GitHub:
|
class GitHub:
|
||||||
|
"""Handles GitHub API requests with caching and response sanitisation."""
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
user_agent: str = (
|
user_agent: str = (
|
||||||
@@ -22,6 +24,12 @@ class GitHub:
|
|||||||
f"https://github.com/bellingcat/octosuite) requests/{requests.__version__}"
|
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.user_agent = user_agent
|
||||||
self.cache = cache
|
self.cache = cache
|
||||||
|
|
||||||
@@ -32,6 +40,16 @@ class GitHub:
|
|||||||
return_response: bool = False,
|
return_response: bool = False,
|
||||||
use_cache: bool = True,
|
use_cache: bool = True,
|
||||||
) -> t.Union[dict, list, Response]:
|
) -> 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:
|
if use_cache and not return_response:
|
||||||
cached = self.cache.get(url, params)
|
cached = self.cache.get(url, params)
|
||||||
if cached is not None:
|
if cached is not None:
|
||||||
@@ -58,7 +76,14 @@ class GitHub:
|
|||||||
def is_valid_entity(
|
def is_valid_entity(
|
||||||
self, _type: t.Literal["user", "org", "repo"], **kwargs
|
self, _type: t.Literal["user", "org", "repo"], **kwargs
|
||||||
) -> bool:
|
) -> 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:
|
try:
|
||||||
type_map = {
|
type_map = {
|
||||||
"user": f"https://api.github.com/users/{kwargs.get('username')}",
|
"user": f"https://api.github.com/users/{kwargs.get('username')}",
|
||||||
@@ -86,6 +111,13 @@ class GitHub:
|
|||||||
return False
|
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]:
|
||||||
|
"""
|
||||||
|
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")
|
pattern = re.compile(r"https://api\.github\.com")
|
||||||
|
|
||||||
if isinstance(response, list):
|
if isinstance(response, list):
|
||||||
|
|||||||
@@ -8,14 +8,25 @@ __all__ = ["User", "Org", "Repo", "Search"]
|
|||||||
|
|
||||||
|
|
||||||
class GitHubEntity:
|
class GitHubEntity:
|
||||||
"""Base class for GitHub entities with common functionality."""
|
"""
|
||||||
|
Base class for GitHub entities with common functionality."""
|
||||||
|
|
||||||
def __init__(self, source: str):
|
def __init__(self, source: str):
|
||||||
|
"""
|
||||||
|
Initialise a GitHub entity.
|
||||||
|
|
||||||
|
:param source: The source identifier for the entity.
|
||||||
|
"""
|
||||||
|
|
||||||
self.endpoint = None
|
self.endpoint = None
|
||||||
self.source = source
|
self.source = source
|
||||||
|
|
||||||
def exists(self) -> tuple[bool, dict]:
|
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
|
# Check cache first
|
||||||
cached = github.cache.get(self.endpoint)
|
cached = github.cache.get(self.endpoint)
|
||||||
if cached is not None:
|
if cached is not None:
|
||||||
@@ -38,58 +49,150 @@ class GitHubEntity:
|
|||||||
|
|
||||||
|
|
||||||
class User(GitHubEntity):
|
class User(GitHubEntity):
|
||||||
|
"""Represents a GitHub user with methods to query user data."""
|
||||||
|
|
||||||
def __init__(self, name: str):
|
def __init__(self, name: str):
|
||||||
|
"""
|
||||||
|
Initialise a User instance.
|
||||||
|
|
||||||
|
:param name: The GitHub username.
|
||||||
|
"""
|
||||||
|
|
||||||
super().__init__(source=name)
|
super().__init__(source=name)
|
||||||
self.name = name
|
self.name = name
|
||||||
self.endpoint = f"{BASE_URL}/users/{name}"
|
self.endpoint = f"{BASE_URL}/users/{name}"
|
||||||
|
|
||||||
def profile(self) -> dict:
|
def profile(self) -> dict:
|
||||||
|
"""
|
||||||
|
Retrieve the user's profile information.
|
||||||
|
|
||||||
|
:return: Dictionary containing user profile data.
|
||||||
|
"""
|
||||||
|
|
||||||
profile = github.get(url=self.endpoint)
|
profile = github.get(url=self.endpoint)
|
||||||
return profile
|
return profile
|
||||||
|
|
||||||
def repos(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
repos = github.get(url=f"{self.endpoint}/repos", params=params)
|
repos = github.get(url=f"{self.endpoint}/repos", params=params)
|
||||||
return repos
|
return repos
|
||||||
|
|
||||||
def subscriptions(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
subscriptions = github.get(url=f"{self.endpoint}/subscriptions", params=params)
|
subscriptions = github.get(url=f"{self.endpoint}/subscriptions", params=params)
|
||||||
return subscriptions
|
return subscriptions
|
||||||
|
|
||||||
def starred(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
starred = github.get(url=f"{self.endpoint}/starred", params=params)
|
starred = github.get(url=f"{self.endpoint}/starred", params=params)
|
||||||
return starred
|
return starred
|
||||||
|
|
||||||
def followers(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
users = github.get(url=f"{self.endpoint}/followers", params=params)
|
users = github.get(url=f"{self.endpoint}/followers", params=params)
|
||||||
return users
|
return users
|
||||||
|
|
||||||
def following(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
users = github.get(url=f"{self.endpoint}/following", params=params)
|
users = github.get(url=f"{self.endpoint}/following", params=params)
|
||||||
return users
|
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:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
orgs = github.get(url=f"{self.endpoint}/orgs", params=params)
|
orgs = github.get(url=f"{self.endpoint}/orgs", params=params)
|
||||||
return orgs
|
return orgs
|
||||||
|
|
||||||
def gists(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
gists = github.get(url=f"{self.endpoint}/gists", params=params)
|
gists = github.get(url=f"{self.endpoint}/gists", params=params)
|
||||||
return gists
|
return gists
|
||||||
|
|
||||||
def events(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
events = github.get(url=f"{self.endpoint}/events", params=params)
|
events = github.get(url=f"{self.endpoint}/events", params=params)
|
||||||
return events
|
return events
|
||||||
|
|
||||||
def received_events(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
received_events = github.get(
|
received_events = github.get(
|
||||||
url=f"{self.endpoint}/received_events", params=params
|
url=f"{self.endpoint}/received_events", params=params
|
||||||
@@ -98,133 +201,337 @@ class User(GitHubEntity):
|
|||||||
|
|
||||||
|
|
||||||
class Org(GitHubEntity):
|
class Org(GitHubEntity):
|
||||||
|
"""Represents a GitHub organisation with methods to query organisation data."""
|
||||||
|
|
||||||
def __init__(self, name: str):
|
def __init__(self, name: str):
|
||||||
|
"""
|
||||||
|
Initialise an Org instance.
|
||||||
|
|
||||||
|
:param name: The GitHub organisation name.
|
||||||
|
"""
|
||||||
|
|
||||||
super().__init__(source=name)
|
super().__init__(source=name)
|
||||||
self.name = name
|
self.name = name
|
||||||
self.endpoint = f"{BASE_URL}/orgs/{name}"
|
self.endpoint = f"{BASE_URL}/orgs/{name}"
|
||||||
|
|
||||||
def profile(self) -> dict:
|
def profile(self) -> dict:
|
||||||
|
"""
|
||||||
|
Retrieve the organisation's profile information.
|
||||||
|
|
||||||
|
:return: Dictionary containing organisation profile data.
|
||||||
|
"""
|
||||||
|
|
||||||
profile = github.get(url=self.endpoint)
|
profile = github.get(url=self.endpoint)
|
||||||
return profile
|
return profile
|
||||||
|
|
||||||
def repos(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
repos = github.get(url=f"{self.endpoint}/repos", params=params)
|
repos = github.get(url=f"{self.endpoint}/repos", params=params)
|
||||||
return repos
|
return repos
|
||||||
|
|
||||||
def events(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
events = github.get(url=f"{self.endpoint}/events", params=params)
|
events = github.get(url=f"{self.endpoint}/events", params=params)
|
||||||
return events
|
return events
|
||||||
|
|
||||||
def hooks(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
hooks = github.get(url=f"{self.endpoint}/hooks", params=params)
|
hooks = github.get(url=f"{self.endpoint}/hooks", params=params)
|
||||||
return hooks
|
return hooks
|
||||||
|
|
||||||
def issues(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
issues = github.get(url=f"{self.endpoint}/issues", params=params)
|
issues = github.get(url=f"{self.endpoint}/issues", params=params)
|
||||||
return issues
|
return issues
|
||||||
|
|
||||||
def members(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
members = github.get(url=f"{self.endpoint}/members", params=params)
|
members = github.get(url=f"{self.endpoint}/members", params=params)
|
||||||
return members
|
return members
|
||||||
|
|
||||||
|
|
||||||
class Repo(GitHubEntity):
|
class Repo(GitHubEntity):
|
||||||
|
"""Represents a GitHub repository with methods to query repository data."""
|
||||||
|
|
||||||
def __init__(self, name: str, owner: str):
|
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}")
|
super().__init__(source=f"{owner}/{name}")
|
||||||
self.name = name
|
self.name = name
|
||||||
self.owner = owner
|
self.owner = owner
|
||||||
self.endpoint = f"{BASE_URL}/repos/{owner}/{name}"
|
self.endpoint = f"{BASE_URL}/repos/{owner}/{name}"
|
||||||
|
|
||||||
def profile(self) -> dict:
|
def profile(self) -> dict:
|
||||||
|
"""
|
||||||
|
Retrieve the repository's information.
|
||||||
|
|
||||||
|
:return: Dictionary containing repository data.
|
||||||
|
"""
|
||||||
|
|
||||||
profile = github.get(url=self.endpoint)
|
profile = github.get(url=self.endpoint)
|
||||||
return profile
|
return profile
|
||||||
|
|
||||||
def forks(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
forks = github.get(url=f"{self.endpoint}/forks", params=params)
|
forks = github.get(url=f"{self.endpoint}/forks", params=params)
|
||||||
return forks
|
return forks
|
||||||
|
|
||||||
def issue_events(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
issue_events = github.get(url=f"{self.endpoint}/issue_events", params=params)
|
issue_events = github.get(url=f"{self.endpoint}/issue_events", params=params)
|
||||||
return issue_events
|
return issue_events
|
||||||
|
|
||||||
def events(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
events = github.get(url=f"{self.endpoint}/events", params=params)
|
events = github.get(url=f"{self.endpoint}/events", params=params)
|
||||||
return events
|
return events
|
||||||
|
|
||||||
def assignees(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
assignees = github.get(url=f"{self.endpoint}/assignees", params=params)
|
assignees = github.get(url=f"{self.endpoint}/assignees", params=params)
|
||||||
return assignees
|
return assignees
|
||||||
|
|
||||||
def branches(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
branches = github.get(url=f"{self.endpoint}/branches", params=params)
|
branches = github.get(url=f"{self.endpoint}/branches", params=params)
|
||||||
return branches
|
return branches
|
||||||
|
|
||||||
def tags(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
tags = github.get(url=f"{self.endpoint}/tags", params=params)
|
tags = github.get(url=f"{self.endpoint}/tags", params=params)
|
||||||
return tags
|
return tags
|
||||||
|
|
||||||
def languages(self) -> dict:
|
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")
|
languages = github.get(url=f"{self.endpoint}/languages")
|
||||||
return languages
|
return languages
|
||||||
|
|
||||||
def stargazers(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
stargazers = github.get(url=f"{self.endpoint}/stargazers", params=params)
|
stargazers = github.get(url=f"{self.endpoint}/stargazers", params=params)
|
||||||
return stargazers
|
return stargazers
|
||||||
|
|
||||||
def subscribers(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
subscribers = github.get(url=f"{self.endpoint}/subscribers", params=params)
|
subscribers = github.get(url=f"{self.endpoint}/subscribers", params=params)
|
||||||
return subscribers
|
return subscribers
|
||||||
|
|
||||||
def commits(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
commits = github.get(url=f"{self.endpoint}/commits", params=params)
|
commits = github.get(url=f"{self.endpoint}/commits", params=params)
|
||||||
return commits
|
return commits
|
||||||
|
|
||||||
def comments(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
comments = github.get(url=f"{self.endpoint}/comments", params=params)
|
comments = github.get(url=f"{self.endpoint}/comments", params=params)
|
||||||
return comments
|
return comments
|
||||||
|
|
||||||
def contents(self, path: str) -> list:
|
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}")
|
contents = github.get(url=f"{self.endpoint}/contents/{path}")
|
||||||
return contents
|
return contents
|
||||||
|
|
||||||
def issues(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
issues = github.get(url=f"{self.endpoint}/issues", params=params)
|
issues = github.get(url=f"{self.endpoint}/issues", params=params)
|
||||||
return issues
|
return issues
|
||||||
|
|
||||||
def releases(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
releases = github.get(url=f"{self.endpoint}/releases", params=params)
|
releases = github.get(url=f"{self.endpoint}/releases", params=params)
|
||||||
return releases
|
return releases
|
||||||
|
|
||||||
def deployments(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
deployments = github.get(url=f"{self.endpoint}/deployments", params=params)
|
deployments = github.get(url=f"{self.endpoint}/deployments", params=params)
|
||||||
return deployments
|
return deployments
|
||||||
|
|
||||||
def labels(self, page: int, per_page: int) -> list:
|
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}
|
params = {"page": page, "per_page": per_page}
|
||||||
labels = github.get(url=f"{self.endpoint}/labels", params=params)
|
labels = github.get(url=f"{self.endpoint}/labels", params=params)
|
||||||
return labels
|
return labels
|
||||||
|
|
||||||
|
|
||||||
class Search:
|
class Search:
|
||||||
|
"""Provides methods to search GitHub for various entity types."""
|
||||||
|
|
||||||
def __init__(self, query: str, page: int, per_page: int):
|
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.query = query
|
||||||
self.page = page
|
self.page = page
|
||||||
self.per_page = per_page
|
self.per_page = per_page
|
||||||
@@ -232,21 +539,51 @@ class Search:
|
|||||||
self.params = {"q": query, "page": page, "per_page": per_page}
|
self.params = {"q": query, "page": page, "per_page": per_page}
|
||||||
|
|
||||||
def repos(self) -> list:
|
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)
|
repos = github.get(url=f"{self.endpoint}/repositories", params=self.params)
|
||||||
return repos
|
return repos
|
||||||
|
|
||||||
def users(self) -> list:
|
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)
|
users = github.get(url=f"{self.endpoint}/users", params=self.params)
|
||||||
return users
|
return users
|
||||||
|
|
||||||
def commits(self) -> list:
|
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)
|
commits = github.get(url=f"{self.endpoint}/commits", params=self.params)
|
||||||
return commits
|
return commits
|
||||||
|
|
||||||
def issues(self) -> list:
|
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)
|
issues = github.get(url=f"{self.endpoint}/issues", params=self.params)
|
||||||
return issues
|
return issues
|
||||||
|
|
||||||
def topics(self) -> list:
|
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)
|
topics = github.get(url=f"{self.endpoint}/topics", params=self.params)
|
||||||
return topics
|
return topics
|
||||||
|
|||||||
@@ -31,6 +31,14 @@ console = Console(log_time=False)
|
|||||||
|
|
||||||
|
|
||||||
def preview_response(data: t.Union[dict, list], source: str, _type: str):
|
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):
|
if isinstance(data, dict):
|
||||||
tree = Tree(
|
tree = Tree(
|
||||||
label=f"\n[bold]{data.get('name') or data.get('login') or data.get('id') or 'Data'}[/bold]",
|
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):
|
elif isinstance(data, list):
|
||||||
preview_data = data[:5]
|
preview_data = data[:5]
|
||||||
tree = Tree(
|
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",
|
guide_style="#444444",
|
||||||
highlight=True,
|
highlight=True,
|
||||||
)
|
)
|
||||||
@@ -74,7 +82,17 @@ def export_response(
|
|||||||
file_formats: list,
|
file_formats: list,
|
||||||
output_dir: str = "../exports",
|
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
|
# Create output directory if it doesn't exist
|
||||||
output_dir = Path(output_dir)
|
output_dir = Path(output_dir)
|
||||||
output_dir.mkdir(exist_ok=True)
|
output_dir.mkdir(exist_ok=True)
|
||||||
@@ -165,6 +183,14 @@ def export_response(
|
|||||||
|
|
||||||
|
|
||||||
def fill_tree(tree: Tree, data: t.Union[dict, list]) -> Tree:
|
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):
|
if isinstance(data, dict):
|
||||||
for key, value in data.items():
|
for key, value in data.items():
|
||||||
if isinstance(value, dict) or isinstance(value, list):
|
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():
|
def check_updates():
|
||||||
|
"""Check for available package updates and display the result."""
|
||||||
|
|
||||||
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__)
|
||||||
@@ -207,6 +235,10 @@ def clear_screen():
|
|||||||
|
|
||||||
|
|
||||||
def ascii_banner(text: str):
|
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()
|
clear_screen()
|
||||||
|
|
||||||
ascii_text = pyfiglet.figlet_format(text=text, font="chunky")
|
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"]):
|
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: str = __pkg__.title()
|
||||||
title += f" | {menu_type.title()}"
|
title += f" | {menu_type.title()}"
|
||||||
console.set_window_title(title=title)
|
console.set_window_title(title=title)
|
||||||
|
|||||||
@@ -26,10 +26,22 @@ __all__ = ["Dialogs"]
|
|||||||
|
|
||||||
|
|
||||||
class Dialogs:
|
class Dialogs:
|
||||||
def __init__(self): ...
|
"""Provides interactive dialogue boxes for user confirmations and information display."""
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
"""Initialise the Dialogs class."""
|
||||||
|
...
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _boolean(title: str, text: str) -> bool:
|
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:
|
try:
|
||||||
result = button_dialog(
|
result = button_dialog(
|
||||||
title=title,
|
title=title,
|
||||||
@@ -41,11 +53,23 @@ class Dialogs:
|
|||||||
return True
|
return True
|
||||||
|
|
||||||
def quit(self) -> bool:
|
def quit(self) -> bool:
|
||||||
|
"""
|
||||||
|
Display a confirmation dialogue for quitting the application.
|
||||||
|
|
||||||
|
:return: True if user confirms quit, False otherwise.
|
||||||
|
"""
|
||||||
|
|
||||||
return self._boolean(
|
return self._boolean(
|
||||||
title="Quit", text="This will close the session, continue?"
|
title="Quit", text="This will close the session, continue?"
|
||||||
)
|
)
|
||||||
|
|
||||||
def clear_cache(self) -> bool:
|
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(
|
return self._boolean(
|
||||||
title="Clear Cache",
|
title="Clear Cache",
|
||||||
text="This will clear all octosuite caches, continue?",
|
text="This will clear all octosuite caches, continue?",
|
||||||
@@ -53,4 +77,5 @@ class Dialogs:
|
|||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def license():
|
def license():
|
||||||
|
"""Display the MIT license notice in a dialogue box."""
|
||||||
message_dialog(title="MIT License", text=LICENSE_NOTICE).run()
|
message_dialog(title="MIT License", text=LICENSE_NOTICE).run()
|
||||||
|
|||||||
@@ -21,7 +21,6 @@ CUSTOM_STYLE = Style(
|
|||||||
|
|
||||||
INSTRUCTIONS = "↑↓ [move] • ⮠ [select]"
|
INSTRUCTIONS = "↑↓ [move] • ⮠ [select]"
|
||||||
EXPORT_INSTRUCTIONS = "↑↓ [move] • ⮠ [confirm] • spacebar [check]"
|
EXPORT_INSTRUCTIONS = "↑↓ [move] • ⮠ [confirm] • spacebar [check]"
|
||||||
POINTER: str = "🖝 "
|
|
||||||
|
|
||||||
dialogs = Dialogs()
|
dialogs = Dialogs()
|
||||||
prompts = Prompts()
|
prompts = Prompts()
|
||||||
@@ -30,9 +29,15 @@ __all__ = ["Menus"]
|
|||||||
|
|
||||||
|
|
||||||
class BaseMenu:
|
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):
|
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
|
# Define which methods require pagination
|
||||||
self.paginated_methods = {
|
self.paginated_methods = {
|
||||||
"repos",
|
"repos",
|
||||||
@@ -76,17 +81,16 @@ class BaseMenu:
|
|||||||
callback: t.Callable,
|
callback: t.Callable,
|
||||||
*callback_args,
|
*callback_args,
|
||||||
) -> bool:
|
) -> 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):
|
if isinstance(instance, User):
|
||||||
target_type = "user"
|
target_type = "user"
|
||||||
elif isinstance(instance, Repo):
|
elif isinstance(instance, Repo):
|
||||||
@@ -95,18 +99,12 @@ class BaseMenu:
|
|||||||
target_type = "org"
|
target_type = "org"
|
||||||
|
|
||||||
with Status(
|
with Status(
|
||||||
f"[dim]Validating {target_type}'s ({identifier}) existence...[/dim]",
|
f"[dim]Validating {target_type} ({identifier})[/dim]...",
|
||||||
console=console,
|
console=console,
|
||||||
) as status:
|
) as status:
|
||||||
exists, response = instance.exists()
|
exists, response = instance.exists()
|
||||||
|
|
||||||
if exists:
|
if not exists:
|
||||||
ascii_banner(text=identifier)
|
|
||||||
console.print(
|
|
||||||
f"[bold][green]✔[/green] {target_type.capitalize()} ({identifier}) exists on GitHub[/bold]"
|
|
||||||
)
|
|
||||||
return True
|
|
||||||
else:
|
|
||||||
status.stop()
|
status.stop()
|
||||||
if response and "message" in response:
|
if response and "message" in response:
|
||||||
console.print(
|
console.print(
|
||||||
@@ -116,6 +114,8 @@ class BaseMenu:
|
|||||||
callback(*callback_args)
|
callback(*callback_args)
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
return True
|
||||||
|
|
||||||
def execute_and_handle_response(
|
def execute_and_handle_response(
|
||||||
self,
|
self,
|
||||||
instance: t.Union[User, Repo, Org, Search],
|
instance: t.Union[User, Repo, Org, Search],
|
||||||
@@ -123,18 +123,19 @@ class BaseMenu:
|
|||||||
target_type: t.Literal["user", "repo", "org"],
|
target_type: t.Literal["user", "repo", "org"],
|
||||||
source: str,
|
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
|
valid_methods = self.paginated_methods | self.non_paginated_methods
|
||||||
if method_name in valid_methods:
|
if method_name in valid_methods:
|
||||||
with Status(
|
with Status(
|
||||||
status=f"[dim]Initialising {target_type} {method_name}...[/dim]",
|
status=f"[dim]Initialising {target_type} {method_name}[/dim]...",
|
||||||
console=console,
|
console=console,
|
||||||
) as status:
|
) as status:
|
||||||
data = self.execute_selection(
|
data = self.execute_selection(
|
||||||
@@ -148,7 +149,14 @@ class BaseMenu:
|
|||||||
self.response_handler(data=data, data_type=method_name, source=source)
|
self.response_handler(data=data, data_type=method_name, source=source)
|
||||||
|
|
||||||
def execute_selection(self, status: Status, **kwargs):
|
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")
|
instance = kwargs.get("instance")
|
||||||
method_name = kwargs.get("method_name")
|
method_name = kwargs.get("method_name")
|
||||||
source = kwargs.get("source")
|
source = kwargs.get("source")
|
||||||
@@ -160,88 +168,111 @@ class BaseMenu:
|
|||||||
prompts.pagination_params() if method_name in self.paginated_methods else {}
|
prompts.pagination_params() if method_name in self.paginated_methods else {}
|
||||||
)
|
)
|
||||||
status.start()
|
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)
|
return method(**params)
|
||||||
|
|
||||||
def response_handler(self, data: t.Union[dict, list], data_type: str, source: str):
|
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:
|
try:
|
||||||
export_choice = q.select(
|
if not data:
|
||||||
"What would you like to do?",
|
console.print(
|
||||||
choices=[
|
f"[bold][yellow]✘[/yellow] No data found for '{source}'[/bold]"
|
||||||
q.Choice(
|
)
|
||||||
title="Export",
|
else:
|
||||||
value="export",
|
preview_response(data=data, source=source, _type=data_type)
|
||||||
description="Export the data",
|
export_choice = q.select(
|
||||||
shortcut_key="e",
|
"What would you like to do?",
|
||||||
),
|
choices=[
|
||||||
q.Choice(
|
q.Choice(
|
||||||
title="Skip",
|
title="Export",
|
||||||
value="skip",
|
value="export",
|
||||||
description="Do nothing, and go back to previous menu",
|
description="Export the data",
|
||||||
shortcut_key="x",
|
shortcut_key="e",
|
||||||
),
|
),
|
||||||
q.Choice(
|
q.Choice(
|
||||||
title="Quit",
|
title="Skip",
|
||||||
value="quit",
|
value="skip",
|
||||||
description="Close this session",
|
description="Do nothing, and go back to previous menu",
|
||||||
),
|
shortcut_key="x",
|
||||||
],
|
),
|
||||||
pointer=POINTER,
|
q.Choice(
|
||||||
style=CUSTOM_STYLE,
|
title="Quit",
|
||||||
instruction=INSTRUCTIONS,
|
value="quit",
|
||||||
).ask()
|
description="Close this session",
|
||||||
|
),
|
||||||
|
],
|
||||||
|
style=CUSTOM_STYLE,
|
||||||
|
instruction=INSTRUCTIONS,
|
||||||
|
).ask()
|
||||||
|
|
||||||
if export_choice == "skip":
|
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,
|
|
||||||
)
|
|
||||||
return
|
return
|
||||||
|
|
||||||
file_formats = q.checkbox(
|
if export_choice == "quit":
|
||||||
"Select export format(s)",
|
if dialogs.quit():
|
||||||
choices=[
|
sys.exit()
|
||||||
q.Choice(
|
else:
|
||||||
title="JSON",
|
clear_screen()
|
||||||
value="json",
|
self.response_handler(
|
||||||
description="Export as JSON file",
|
data=data,
|
||||||
),
|
data_type=data_type,
|
||||||
q.Choice(
|
source=source,
|
||||||
title="CSV",
|
)
|
||||||
value="csv",
|
return
|
||||||
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()
|
|
||||||
|
|
||||||
export_response(
|
file_formats = q.checkbox(
|
||||||
data=data, data_type=data_type, source=source, file_formats=file_formats
|
"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:
|
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):
|
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 = {
|
navigation_handlers = {
|
||||||
"back": lambda: self.main_menu(),
|
"back": lambda: self.main_menu(),
|
||||||
"quit": lambda: (
|
"quit": lambda: (
|
||||||
@@ -257,7 +288,11 @@ class BaseMenu:
|
|||||||
|
|
||||||
|
|
||||||
class Menus(BaseMenu):
|
class Menus(BaseMenu):
|
||||||
|
"""Main menu system providing interactive interfaces for GitHub data queries."""
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
|
"""Initialise the Menus class with mode handlers for different query types."""
|
||||||
|
|
||||||
super().__init__(main_menu=self.main)
|
super().__init__(main_menu=self.main)
|
||||||
|
|
||||||
self.mode_handlers = {
|
self.mode_handlers = {
|
||||||
@@ -268,7 +303,8 @@ class Menus(BaseMenu):
|
|||||||
}
|
}
|
||||||
|
|
||||||
def main(self):
|
def main(self):
|
||||||
"""Main menu to select mode."""
|
"""Display the main menu for selecting query mode."""
|
||||||
|
|
||||||
set_menu_title(menu_type="home")
|
set_menu_title(menu_type="home")
|
||||||
clear_screen()
|
clear_screen()
|
||||||
try:
|
try:
|
||||||
@@ -318,7 +354,6 @@ class Menus(BaseMenu):
|
|||||||
description="Close this session",
|
description="Close this session",
|
||||||
),
|
),
|
||||||
],
|
],
|
||||||
pointer=POINTER,
|
|
||||||
style=CUSTOM_STYLE,
|
style=CUSTOM_STYLE,
|
||||||
instruction=INSTRUCTIONS,
|
instruction=INSTRUCTIONS,
|
||||||
).ask()
|
).ask()
|
||||||
@@ -350,7 +385,12 @@ class Menus(BaseMenu):
|
|||||||
sys.exit()
|
sys.exit()
|
||||||
|
|
||||||
def search(self, query: t.Optional[str] = None):
|
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")
|
set_menu_title(menu_type="search")
|
||||||
clear_screen()
|
clear_screen()
|
||||||
if query is None:
|
if query is None:
|
||||||
@@ -409,7 +449,6 @@ class Menus(BaseMenu):
|
|||||||
shortcut_key="q",
|
shortcut_key="q",
|
||||||
),
|
),
|
||||||
],
|
],
|
||||||
pointer=POINTER,
|
|
||||||
style=CUSTOM_STYLE,
|
style=CUSTOM_STYLE,
|
||||||
instruction=INSTRUCTIONS,
|
instruction=INSTRUCTIONS,
|
||||||
use_shortcuts=True,
|
use_shortcuts=True,
|
||||||
@@ -431,7 +470,7 @@ class Menus(BaseMenu):
|
|||||||
# 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 Status(
|
with Status(
|
||||||
status=f"[dim]Initialising {option} search...[/dim]", console=console
|
status=f"[dim]Initialising {option} search[/dim]...", console=console
|
||||||
) as status:
|
) as status:
|
||||||
status.stop()
|
status.stop()
|
||||||
params = prompts.pagination_params()
|
params = prompts.pagination_params()
|
||||||
@@ -445,7 +484,7 @@ class Menus(BaseMenu):
|
|||||||
)
|
)
|
||||||
|
|
||||||
method = getattr(search, option)
|
method = getattr(search, option)
|
||||||
status.update(f"[dim]Searching {option} for {query}...[/dim]")
|
status.update(f"[dim]Searching {option} for {query}[/dim]...")
|
||||||
data = method()
|
data = method()
|
||||||
|
|
||||||
if data:
|
if data:
|
||||||
@@ -461,7 +500,13 @@ class Menus(BaseMenu):
|
|||||||
self.search(query=query)
|
self.search(query=query)
|
||||||
|
|
||||||
def user(self, username: t.Optional[str] = None, is_validated: bool = False):
|
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")
|
set_menu_title(menu_type="user")
|
||||||
clear_screen()
|
clear_screen()
|
||||||
if username is None:
|
if username is None:
|
||||||
@@ -475,14 +520,14 @@ class Menus(BaseMenu):
|
|||||||
|
|
||||||
user = User(name=username)
|
user = User(name=username)
|
||||||
|
|
||||||
if not is_validated:
|
if not self.target_validator(
|
||||||
if not self.target_validator(
|
identifier=username,
|
||||||
identifier=username,
|
instance=user,
|
||||||
instance=user,
|
callback=self.user,
|
||||||
callback=self.user,
|
):
|
||||||
):
|
return
|
||||||
return
|
|
||||||
|
|
||||||
|
ascii_banner(text=username)
|
||||||
option = q.select(
|
option = q.select(
|
||||||
"What would you like to do/get?",
|
"What would you like to do/get?",
|
||||||
choices=[
|
choices=[
|
||||||
@@ -555,7 +600,6 @@ class Menus(BaseMenu):
|
|||||||
shortcut_key="q",
|
shortcut_key="q",
|
||||||
),
|
),
|
||||||
],
|
],
|
||||||
pointer=POINTER,
|
|
||||||
style=CUSTOM_STYLE,
|
style=CUSTOM_STYLE,
|
||||||
instruction=INSTRUCTIONS,
|
instruction=INSTRUCTIONS,
|
||||||
use_shortcuts=True,
|
use_shortcuts=True,
|
||||||
@@ -566,32 +610,37 @@ class Menus(BaseMenu):
|
|||||||
return
|
return
|
||||||
|
|
||||||
# Handle navigation
|
# Handle navigation
|
||||||
if self.navigation_handler(option, self.user, username):
|
elif self.navigation_handler(option, self.user, username):
|
||||||
return
|
return
|
||||||
|
|
||||||
# Handle change username
|
# Handle change username
|
||||||
if option == "change_username":
|
elif option == "change_username":
|
||||||
self.user()
|
self.user()
|
||||||
return
|
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
|
# After handling response, show menu again WITHOUT re-validating
|
||||||
self.execute_and_handle_response(
|
self.user(username=username, is_validated=True)
|
||||||
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)
|
|
||||||
|
|
||||||
def repo(
|
def repo(
|
||||||
self,
|
self,
|
||||||
name: t.Optional[str] = None,
|
name: t.Optional[str] = None,
|
||||||
owner: 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")
|
set_menu_title(menu_type="repo")
|
||||||
clear_screen()
|
clear_screen()
|
||||||
if name is None or owner is None:
|
if name is None or owner is None:
|
||||||
@@ -614,14 +663,14 @@ class Menus(BaseMenu):
|
|||||||
source = f"{owner}/{name}"
|
source = f"{owner}/{name}"
|
||||||
|
|
||||||
# Only validate if not already validated
|
# Only validate if not already validated
|
||||||
if not is_validated:
|
if not self.target_validator(
|
||||||
if not self.target_validator(
|
identifier=source,
|
||||||
identifier=source,
|
instance=repo,
|
||||||
instance=repo,
|
callback=self.repo,
|
||||||
callback=self.repo,
|
):
|
||||||
):
|
return
|
||||||
return
|
|
||||||
|
|
||||||
|
ascii_banner(text=source)
|
||||||
option = q.select(
|
option = q.select(
|
||||||
"What would you like to do/get?",
|
"What would you like to do/get?",
|
||||||
choices=[
|
choices=[
|
||||||
@@ -733,20 +782,11 @@ class Menus(BaseMenu):
|
|||||||
shortcut_key="q",
|
shortcut_key="q",
|
||||||
),
|
),
|
||||||
],
|
],
|
||||||
pointer=POINTER,
|
|
||||||
style=CUSTOM_STYLE,
|
style=CUSTOM_STYLE,
|
||||||
instruction=INSTRUCTIONS,
|
instruction=INSTRUCTIONS,
|
||||||
use_shortcuts=True,
|
use_shortcuts=True,
|
||||||
).ask()
|
).ask()
|
||||||
|
|
||||||
if option is None:
|
|
||||||
self.main()
|
|
||||||
return
|
|
||||||
|
|
||||||
# Handle navigation
|
|
||||||
if self.navigation_handler(option, self.repo, name, owner):
|
|
||||||
return
|
|
||||||
|
|
||||||
# Handle change options
|
# Handle change options
|
||||||
change_handlers = {
|
change_handlers = {
|
||||||
"change_repo_name": lambda: self.repo(owner=owner),
|
"change_repo_name": lambda: self.repo(owner=owner),
|
||||||
@@ -754,23 +794,36 @@ class Menus(BaseMenu):
|
|||||||
"change_both": lambda: self.repo(),
|
"change_both": lambda: self.repo(),
|
||||||
}
|
}
|
||||||
|
|
||||||
if option in change_handlers:
|
if option is None:
|
||||||
change_handlers[option]()
|
self.main()
|
||||||
return
|
return
|
||||||
|
|
||||||
# Execute action and handle response
|
# Handle navigation
|
||||||
self.execute_and_handle_response(
|
elif self.navigation_handler(option, self.repo, name, owner):
|
||||||
instance=repo,
|
return
|
||||||
method_name=option,
|
|
||||||
target_type="repo",
|
|
||||||
source=source,
|
|
||||||
)
|
|
||||||
|
|
||||||
# After handling response, show menu again WITHOUT re-validating
|
elif option in change_handlers:
|
||||||
self.repo(name=name, owner=owner, is_validated=True)
|
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")
|
set_menu_title(menu_type="org")
|
||||||
clear_screen()
|
clear_screen()
|
||||||
if name is None:
|
if name is None:
|
||||||
@@ -784,14 +837,12 @@ class Menus(BaseMenu):
|
|||||||
|
|
||||||
org = Org(name=name)
|
org = Org(name=name)
|
||||||
|
|
||||||
# Only validate if not already validated
|
if not self.target_validator(
|
||||||
if not is_validated:
|
identifier=name,
|
||||||
if not self.target_validator(
|
instance=org,
|
||||||
identifier=name,
|
callback=self.org,
|
||||||
instance=org,
|
):
|
||||||
callback=self.org,
|
return
|
||||||
):
|
|
||||||
return
|
|
||||||
|
|
||||||
option = q.select(
|
option = q.select(
|
||||||
"What would you like to do?",
|
"What would you like to do?",
|
||||||
@@ -844,7 +895,6 @@ class Menus(BaseMenu):
|
|||||||
shortcut_key="q",
|
shortcut_key="q",
|
||||||
),
|
),
|
||||||
],
|
],
|
||||||
pointer=POINTER,
|
|
||||||
style=CUSTOM_STYLE,
|
style=CUSTOM_STYLE,
|
||||||
instruction=INSTRUCTIONS,
|
instruction=INSTRUCTIONS,
|
||||||
use_shortcuts=True,
|
use_shortcuts=True,
|
||||||
@@ -855,18 +905,19 @@ class Menus(BaseMenu):
|
|||||||
return
|
return
|
||||||
|
|
||||||
# Handle navigation
|
# Handle navigation
|
||||||
if self.navigation_handler(option, self.org, name):
|
elif self.navigation_handler(option, self.org, name):
|
||||||
return
|
return
|
||||||
|
|
||||||
# Handle change org
|
# Handle change org
|
||||||
if option == "change_org":
|
elif option == "change_org":
|
||||||
self.org()
|
self.org()
|
||||||
return
|
return
|
||||||
|
|
||||||
# Execute action and handle response
|
else:
|
||||||
self.execute_and_handle_response(
|
# Execute action and handle response
|
||||||
instance=org, method_name=option, target_type="org", source=name
|
self.execute_and_handle_response(
|
||||||
)
|
instance=org, method_name=option, target_type="org", source=name
|
||||||
|
)
|
||||||
|
|
||||||
# After handling response, show menu again WITHOUT re-validating
|
# After handling response, show menu again WITHOUT re-validating
|
||||||
self.org(name=name, is_validated=True)
|
self.org(name=name)
|
||||||
|
|||||||
@@ -7,7 +7,10 @@ __all__ = ["Prompts"]
|
|||||||
|
|
||||||
|
|
||||||
class Prompts:
|
class Prompts:
|
||||||
|
"""Provides interactive prompts for user input collection."""
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
|
"""Initialise the Prompts class."""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
@@ -17,17 +20,40 @@ class Prompts:
|
|||||||
style: t.Optional[Style] = None,
|
style: t.Optional[Style] = None,
|
||||||
qmark: t.Optional[str] = "?",
|
qmark: t.Optional[str] = "?",
|
||||||
) -> 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,
|
message=message,
|
||||||
instruction=instruction,
|
instruction=instruction,
|
||||||
style=style,
|
style=style,
|
||||||
qmark=qmark,
|
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()
|
).ask()
|
||||||
|
|
||||||
|
if result is None:
|
||||||
|
raise KeyboardInterrupt
|
||||||
|
|
||||||
|
return result
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def pagination_params() -> dict:
|
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:
|
try:
|
||||||
page = q.text(message="Page", default="1", qmark="n").ask()
|
page = q.text(message="Page", default="1", qmark="n").ask()
|
||||||
per_page = q.text(
|
per_page = q.text(
|
||||||
|
|||||||
Reference in New Issue
Block a user