Update update_unstable.py

- Port to Python 3.8 syntax
- Use type annotations
- Add a matrix notification using matrix-commander

obs/update_unstable.py

@@ -3,43 +3,56 @@
 # SPDX-License-Identifier: BSD-3-clause
 
 import argparse
+from datetime import date
+from collections import defaultdict
 import logging
 import json
 from pathlib import Path
+from typing import Union, Dict, Any
 from urllib.parse import quote
 
 import git
 import requests
+import sarge
+
 
 API_URL = "https://invent.kde.org/api/v4/projects/"
 OBS_URL = "https://api.opensuse.org/trigger/runservice"
 
+MESSAGE_TEMPLATE = f"""
+### OBS package update complete
+
+Stats for {date.today().strftime('%Y-%m-%d')}:
+
+"""
+
 
 class GitHashCache:
 
-    def __init__(self, cache_file):
+    def __init__(self, cache_file: str) -> None:
         self.cache = cache_file
-        self._data = dict()
+        self._data: Dict[str, Dict[str, str]] = dict()
 
-    def __getitem__(self, key):
+    def __getitem__(self, key: str) -> Dict[str, str]:
 
         if key not in self._data:
-            return ""
+            raise KeyError
 
         return self._data[key]
 
-    def __setitem__(self, key, value):
+    def __setitem__(self, key: str, value: Dict[str, str]) -> None:
         self._data[key] = value
 
-    def get(self, key, *args, **kwargs):
+    def get(self, key: str,
+            *args: Any, **kwargs: Any) -> Union[None, str, Dict[str, str]]:
         return self._data.get(key, *args, **kwargs)
 
-    def save(self):
+    def save(self) -> None:
         logging.debug("Saving pickled data")
         with open(self.cache, "w") as handle:
             json.dump(self._data, handle, indent=4)
 
-    def load(self):
+    def load(self) -> None:
 
         if not Path(self.cache).exists():
             logging.debug("File cache not found, not loading")
@@ -49,7 +62,7 @@ class GitHashCache:
             self._data = json.load(handle)
 
 
-def project_exists(project):
+def project_exists(project: str) -> bool:
     # We want / to get quoted, so put safe to ""
     project_name = quote(project, safe="")
     request = requests.get(API_URL + project_name)
@@ -60,7 +73,7 @@ def project_exists(project):
     return False
 
 
-def lsremote(url):
+def lsremote(url: str) -> Dict[str, str]:
 
     remote_refs = {}
     gitcmd = git.cmd.Git()
@@ -71,12 +84,13 @@ def lsremote(url):
     return remote_refs
 
 
-def get_remote_hash(url, branch="master"):
+def get_remote_hash(url: str, branch: str = "master") -> str:
     refs = "refs/heads/{}".format(branch)
     return lsremote(url)[refs]
 
 
-def trigger_update(repository, package_name, token):
+def trigger_update(repository: str, package_name: str,
+                   token: str) -> Union[requests.Response, bool]:
 
     header = {"Authorization": f"Token {token}"}
     parameters = {"project": repository, "package": package_name}
@@ -94,9 +108,11 @@ def trigger_update(repository, package_name, token):
     return result
 
 
-def update_package(hash_data, package_name,
-                   remote_name, obs_repository,
-                   branch, token):
+def update_package(hash_data: GitHashCache, package_name: str,
+                   remote_name: str, obs_repository: str,
+                   branch: str,
+                   token: str,
+                   stats: Dict[str, int]) -> None:
 
     repo_name = "https://invent.kde.org/{}".format(remote_name)
 
@@ -119,19 +135,26 @@ def update_package(hash_data, package_name,
         logging.debug("Hash doesn't match, updating")
         if trigger_update(obs_repository, package_name, token):
             hash_data[obs_repository][remote_name] = remote_hash
+            stats["updated"] += 1
             hash_data.save()
+        else:
+            stats["errors"] += 1
 
 
-def update_packages(cache_file, repo_mapping_file, token):
+def update_packages(cache_file: str,
+                    repo_mapping_file: str, token: str) -> None:
 
     hash_data = GitHashCache(cache_file)
     hash_data.load()
 
+    stats = dict()
+
     with open(repo_mapping_file, "r") as mapping:
 
         repo_data = json.load(mapping)
 
         for obs_repository, branch_data in repo_data.items():
+            repo_stats: Dict[str, int] = defaultdict(int)
 
             logging.info("Updating packages for %s", obs_repository)
 
@@ -144,13 +167,30 @@ def update_packages(cache_file, repo_mapping_file, token):
                               package_name, obs_name)
                 logging.debug("Using branch %s", branch)
                 update_package(hash_data, obs_name, kde_name, obs_repository,
-                               branch, token)
+                               branch, token, repo_stats)
+            stats[obs_repository] = repo_stats
 
     logging.debug("Saving data")
     hash_data.save()
+    notify_matrix(stats)
 
 
-def commit_changes(cache_file, repo_home):
+def notify_matrix(stats: Dict[str, Dict[str, int]]) -> None:
+
+    structure = [MESSAGE_TEMPLATE]
+    for key, value in stats.items():
+        row = (f"* {key}: {stats['updated']} updated packages,"
+               f" {stats['errors']} errors")
+        structure.append(row)
+
+    message = "\n".join(structure)
+
+    cmd = ["matrix_commander.py", "--markdown", "-m", message]
+    logging.debug("Sending Matrix notification")
+    sarge.run(cmd)
+
+
+def commit_changes(cache_file: str, repo_home: str) -> None:
 
     repo = git.Repo(repo_home)
     repo.index.add([cache_file])
@@ -160,7 +200,7 @@ def commit_changes(cache_file, repo_home):
     origin.push()
 
 
-def main():
+def main() -> None:
 
     parser = argparse.ArgumentParser()
     parser.add_argument(