coroutine cancellation added

sources/graphics_chart_drawer.py

@@ -8,6 +8,7 @@ from manager_download import DownloadManager as DM
 
 
 MAX_LANGUAGES = 5
+GRAPH_PATH = "assets/bar_graph.png"
 
 
 async def create_loc_graph(yearly_data: Dict, save_path: str):

sources/graphics_list_formatter.py

@@ -34,18 +34,18 @@ def make_graph(percent: float):
     return f"{done_block * percent_quart}{empty_block * (25 - percent_quart)}"
 
 
-def make_list(data: Dict = None, names: List[str] = None, texts: List[str] = None, percents: List[float] = None, top_num: int = 5, sort: bool = True) -> str:
+def make_list(data: List = None, names: List[str] = None, texts: List[str] = None, percents: List[float] = None, top_num: int = 5, sort: bool = True) -> str:
     """
     Make List
     """
     if data is not None:
-        names = [value for key, value in data if key == "name"] if names is None else names
-        texts = [value for key, value in data if key == "text"] if texts is None else texts
-        percents = [value for key, value in data if key == "percent"] if percents is None else percents
+        names = [value for item in data for key, value in item.items() if key == "name"] if names is None else names
+        texts = [value for item in data for key, value in item.items() if key == "text"] if texts is None else texts
+        percents = [value for item in data for key, value in item.items() if key == "percent"] if percents is None else percents
 
     data = list(zip(names, texts, percents))
-    top_data = sorted(data[:top_num], key=lambda _, __, p: p) if sort else data[:top_num]
-    data_list = [f"{n:25]}{' ' * (25 - len(n))}{t}{' ' * (20 - len(t))}{make_graph(p)}   {p:05.2f} % " for n, t, p in top_data]
+    top_data = sorted(data[:top_num], key=lambda record: record[2]) if sort else data[:top_num]
+    data_list = [f"{n[:25]}{' ' * (25 - len(n))}{t}{' ' * (20 - len(t))}{make_graph(p)}   {p:05.2f} % " for n, t, p in top_data]
     return '\n'.join(data_list)
 
 
@@ -73,25 +73,25 @@ async def make_commit_day_time_list(time_zone: str) -> str:
     sum_week = sum(week_days)
     day_times = day_times[1:] + day_times[:1]
 
-    day_time_names = [f"{DAY_TIME_EMOJI[i]} {LM.t(DAY_TIME_NAMES[i])}" for i in range(len(day_times))]
-    day_time_texts = [f'{day_time} commits' for day_time in day_times]
-    day_time_percents = [round((day_time / sum_day) * 100, 2) for day_time in day_times]
+    dt_names = [f"{DAY_TIME_EMOJI[i]} {LM.t(DAY_TIME_NAMES[i])}" for i in range(len(day_times))]
+    dt_texts = [f'{day_time} commits' for day_time in day_times]
+    dt_percents = [round((day_time / sum_day) * 100, 2) for day_time in day_times]
     title = LM.t("I am an Early") if sum(day_times[0:2]) >= sum(day_times[2:4]) else LM.t("I am a Night")
-    stats += f"**{title}** \n\n```text\n{make_list(names=day_time_names, texts=day_time_texts, percents=day_time_percents, top_num=7)}\n\n```\n"
+    stats += f"**{title}** \n\n```text\n{make_list(names=dt_names, texts=dt_texts, percents=dt_percents, top_num=7, sort=False)}\n```\n"
 
     if EM.SHOW_DAYS_OF_WEEK:
-        week_day_names = [LM.t(week_day) for week_day in WEEK_DAY_NAMES]
-        week_day_texts = [f'{week_day} commits' for week_day in week_days]
-        week_day_percents = [round((week_day / sum_week) * 100, 2) for week_day in week_days]
-        title = LM.t("I am Most Productive on") % week_day_names[week_day_percents.index(max(week_day_percents))]
-        stats += f"📅 **{title}** \n\n```text\n{make_list(names=week_day_names, texts=week_day_texts, percents=week_day_percents, top_num=7)}\n\n```\n"
+        wd_names = [LM.t(week_day) for week_day in WEEK_DAY_NAMES]
+        wd_texts = [f'{week_day} commits' for week_day in week_days]
+        wd_percents = [round((week_day / sum_week) * 100, 2) for week_day in week_days]
+        title = LM.t("I am Most Productive on") % wd_names[wd_percents.index(max(wd_percents))]
+        stats += f"📅 **{title}** \n\n```text\n{make_list(names=wd_names, texts=wd_texts, percents=wd_percents, top_num=7, sort=False)}\n```\n"
 
     return stats
 
 
-def make_language_per_repo_list(result: Dict) -> str:
+def make_language_per_repo_list(repositories: Dict) -> str:
     language_count = dict()
-    repos_with_language = [repo for repo in result["data"]["user"]["repositories"]["edges"] if repo["node"]["primaryLanguage"] is not None]
+    repos_with_language = [repo for repo in repositories["data"]["user"]["repositories"]["edges"] if repo["node"]["primaryLanguage"] is not None]
     for repo in repos_with_language:
         language = repo["node"]["primaryLanguage"]["name"]
         language_count[language] = language_count.get(language, {"count": 0})

sources/main.py

@@ -7,12 +7,12 @@ from urllib.parse import quote
 
 from humanize import intword, naturalsize, intcomma, precisedelta
 
-from manager_download import init_download_manager, DownloadManager as DM
+from manager_download import init_download_manager, DownloadManager as DM, close_download_manager
 from manager_environment import EnvironmentManager as EM
 from manager_github import init_github_manager, GitHubManager as GHM
 from manager_localization import init_localization_manager, LocalizationManager as LM
-from graphics_chart_drawer import create_loc_graph
-from yearly_commit_calculator import GRAPH_PATH, calculate_yearly_commit_data
+from graphics_chart_drawer import create_loc_graph, GRAPH_PATH
+from yearly_commit_calculator import calculate_yearly_commit_data
 from graphics_list_formatter import make_list, make_commit_day_time_list, make_language_per_repo_list
 
 
@@ -47,7 +47,7 @@ async def get_waka_time_stats() -> str:
             os_list = no_activity if len(data["data"]["operating_systems"]) == 0 else make_list(data["data"]["operating_systems"])
             stats += f"💻 {LM.t('operating system')}: \n{os_list}\n\n"
 
-        stats += '```\n\n'
+        stats = f"{stats[:-1]}```\n\n"
 
     return stats
 
@@ -141,6 +141,7 @@ async def main():
 
     if GHM.update_readme(await get_stats()):
         print("Readme updated!")
+    await close_download_manager()
 
 
 if __name__ == '__main__':
@@ -148,8 +149,3 @@ if __name__ == '__main__':
     run(main())
     run_delta = datetime.now() - start_time
     print(f"Program processed in {precisedelta(run_delta, minimum_unit='microseconds')}.")
-
-# TODO: check function and variable naming
-# TODO: check type hints
-# TODO: sorted to max / min
-# TODO: drop not awaited coroutines

sources/manager_download.py

@@ -110,6 +110,15 @@ async def init_download_manager():
     })
 
 
+async def close_download_manager():
+    """
+    Initialize download manager:
+    - Setup headers for GitHub GraphQL requests.
+    - Launch static queries in background.
+    """
+    await DownloadManager.close_remote_resources("linguist", "waka_latest", "waka_all", "github_stats")
+
+
 class DownloadManager:
     """
     Class for handling and caching all kinds of requests.
@@ -135,6 +144,16 @@ class DownloadManager:
             DownloadManager._REMOTE_RESOURCES_CACHE[resource] = DownloadManager._client.get(url)
         DownloadManager._client.headers = github_headers
 
+    @staticmethod
+    async def close_remote_resources(*resource: str):
+        """
+        Prepare DownloadManager to launch GitHub API queries and launch all static queries.
+        :param resources: Dictionary of static queries, "IDENTIFIER": "URL".
+        :param github_headers: Dictionary of headers for GitHub API queries.
+        """
+        for resource in [DownloadManager._REMOTE_RESOURCES_CACHE[r] for r in resource if isinstance(DownloadManager._REMOTE_RESOURCES_CACHE[r], Awaitable)]:
+            resource.cancel()
+
     @staticmethod
     async def _get_remote_resource(resource: str, convertor: Optional[Callable[[bytes], Dict]]) -> Dict:
         """

sources/yearly_commit_calculator.py

@@ -7,13 +7,10 @@ from manager_environment import EnvironmentManager as EM
 from manager_github import GitHubManager as GHM
 
 
-GRAPH_PATH = "assets/bar_graph.png"
-
-
-async def calculate_yearly_commit_data(repository_data: Dict) -> Dict:
+async def calculate_yearly_commit_data(repositories: Dict) -> Dict:
     yearly_data = dict()
-    total = len(repository_data["data"]["user"]["repositories"]["edges"])
-    for ind, repo in enumerate(repository_data["data"]["user"]["repositories"]["edges"]):
+    total = len(repositories["data"]["user"]["repositories"]["edges"])
+    for ind, repo in enumerate(repositories["data"]["user"]["repositories"]["edges"]):
         if repo["node"]["name"] not in EM.IGNORED_REPOS:
             print(f"{ind + 1}/{total}", "Retrieving repo:", repo["node"]["owner"]["login"], repo["node"]["name"])
             await update_yearly_data_with_commit_stats(repo["node"], yearly_data)