add github stars stat

Author: veghdev

src/writegithubstat/githubstat.py

@@ -1,16 +1,19 @@
 from abc import ABC, abstractmethod
 import os
+from pathlib import Path
 import logging
 import requests
 from datetime import date
 
 import pandas as pd
 
 
-class _GithubAuth:
+class GithubAuth:
     def __init__(self, owner: str, repo: str, token: str) -> None:
-        self._header = _GithubAuth._get_auth_header(token)
-        self._repo_id = _GithubAuth._get_repo_id(owner, repo, self._header)
+        self._owner = owner
+        self._repo = repo
+        self._header = GithubAuth._get_auth_header(token)
+        self._repo_id = GithubAuth._get_repo_id(self._owner, self._repo, self._header)
 
     @staticmethod
     def _get_auth_header(token: str) -> dict:
@@ -29,42 +32,33 @@ def _get_repo_id(owner: str, repo: str, auth_header: str) -> str:
             return repository["id"]
         else:
             raise requests.HTTPError(
-                f"Request failed with status code {response.status_code}"
+                f"Request failed with status code {response.status_code}: {response.text}"
             )
 
     @property
-    def header(self) -> dict:
-        return self._header
+    def owner(self) -> str:
+        return self._owner
+
+    @property
+    def repo(self) -> str:
+        return self._repo
 
     @property
     def repo_id(self) -> str:
         return self._repo_id
 
+    @property
+    def header(self) -> dict:
+        return self._header
 
-class _GithubStat:
-    @staticmethod
-    def _dict_to_dataframe(data) -> pd.DataFrame:
-        df = pd.DataFrame(data)
-        return df
-
-    @staticmethod
-    def _get_stat(auth: _GithubAuth, path: str) -> pd.DataFrame:
-        url = f"https://api.github.com/repositories/{auth.repo_id}/{path}"
-        response = requests.get(url, headers=auth.header)
-        if response.status_code == 200:
-            raw_data = response.json()
-            data = _GithubStat._dict_to_dataframe(raw_data)
-            return data
-        else:
-            raise requests.HTTPError(
-                f"Request failed with status code {response.status_code}"
-            )
 
+class GithubStatType(ABC):
+    def __init__(self, auth: GithubAuth) -> None:
+        self._auth = auth
 
-class _GithubStatType(ABC):
     @property
     @abstractmethod
-    def path(self):
+    def url(self):
         pass
 
     @property
@@ -77,16 +71,15 @@ def dimensions(self):
     def measures(self):
         pass
 
-    @property
-    @abstractmethod
-    def name(self):
+    @staticmethod
+    def process_stat(data):
         pass
 
 
-class Referrers(_GithubStatType):
+class Referrers(GithubStatType):
     @property
-    def path(self):
-        return "traffic/popular/referrers"
+    def url(self):
+        return f"https://api.github.com/repositories/{self._auth.repo_id}/traffic/popular/referrers"
 
     @property
     def dimensions(self):
@@ -96,15 +89,16 @@ def dimensions(self):
     def measures(self):
         return ["count", "uniques"]
 
-    @property
-    def name(self):
-        return "referrers"
+    @staticmethod
+    def process_stat(data):
+        df = pd.DataFrame(data)
+        return df
 
 
-class Paths(_GithubStatType):
+class Paths(GithubStatType):
     @property
-    def path(self):
-        return "traffic/popular/paths"
+    def url(self):
+        return f"https://api.github.com/repositories/{self._auth.repo_id}/traffic/popular/paths"
 
     @property
     def dimensions(self):
@@ -114,34 +108,63 @@ def dimensions(self):
     def measures(self):
         return ["count", "uniques"]
 
+    @staticmethod
+    def process_stat(data):
+        df = pd.DataFrame(data)
+        return df
+
+
+class Stars(GithubStatType):
+    @property
+    def url(self):
+        return f"https://api.github.com/repos/{self._auth.owner}/{self._auth.repo}"
+
     @property
-    def name(self):
-        return "paths"
+    def dimensions(self):
+        return []
+
+    @property
+    def measures(self):
+        return ["stars_cumulative"]
+
+    @staticmethod
+    def process_stat(data):
+        stars = data["stargazers_count"]
+        df = pd.DataFrame({"stars_cumulative": [stars]})
+        return df
+
+
+class _GithubStat:
+    @staticmethod
+    def _get_stat(stat_type: GithubStatType, auth_header: dict) -> pd.DataFrame:
+        response = requests.get(stat_type.url, headers=auth_header)
+        if response.status_code == 200:
+            return stat_type.process_stat(response.json())
+        else:
+            raise requests.HTTPError(
+                f"Request failed with status code {response.status_code}: {response.text}"
+            )
 
 
 class WriteGithubStat:
-    def __init__(self, owner: str, repo: str, token: str) -> None:
+    def __init__(self, auth: GithubAuth) -> None:
         self._date = date.today().strftime("%Y-%m-%d")
-        self._owner = owner
-        self._repo = repo
-        self._auth = _GithubAuth(self._owner, self._repo, token)
+        self._auth = auth
 
-    def write_stats(self, outdir: str, prefix: str) -> None:
-        os.makedirs(outdir, exist_ok=True)
-        for stat_type in Referrers(), Paths():
-            self._write_stat(stat_type, outdir, prefix)
+    @property
+    def date(self) -> str:
+        return self._date
 
-    def _write_stat(self, stat_type: _GithubStatType, outdir: str, prefix: str) -> None:
-        year = self._date[0:4]
-        csv = f"{outdir}/{year}_{prefix}_githubstat_{stat_type.name}.csv"
+    def write_stat(self, stat_type: GithubStatType, csv: Path) -> None:
+        os.makedirs(csv.parent, exist_ok=True)
         stats = self._get_stats(stat_type)
         logging.info(stats)
         stored_stats = self._get_stored_stats(csv)
         merged_stats = self._merge_stats(stored_stats, stats)
         merged_stats.to_csv(csv, index=False)
 
-    def _get_stats(self, stat_type: _GithubStatType) -> pd.DataFrame:
-        stat = _GithubStat._get_stat(self._auth, stat_type.path)
+    def _get_stats(self, stat_type: GithubStatType) -> pd.DataFrame:
+        stat = _GithubStat._get_stat(stat_type, self._auth.header)
         if stat.empty:
             empty = {
                 **{col: "-" for col in stat_type.dimensions},
@@ -153,8 +176,8 @@ def _get_stats(self, stat_type: _GithubStatType) -> pd.DataFrame:
 
     def _insert_metadata(self, df: pd.DataFrame) -> pd.DataFrame:
         df.insert(0, "date", self._date)
-        df.insert(1, "owner", self._owner)
-        df.insert(2, "repo", self._repo)
+        df.insert(1, "owner", self._auth.owner)
+        df.insert(2, "repo", self._auth.repo)
         return df
 
     def _get_stored_stats(self, path: str) -> pd.DataFrame:
@@ -171,8 +194,8 @@ def _merge_stats(
             stored_stats = stored_stats.drop(
                 stored_stats[
                     (stored_stats["date"] == self._date)
-                    & (stored_stats["owner"] == self._owner)
-                    & (stored_stats["repo"] == self._repo)
+                    & (stored_stats["owner"] == self._auth.owner)
+                    & (stored_stats["repo"] == self._auth.repo)
                 ].index
             )
             stats = pd.concat([stored_stats, stats])