Optimize GitHub handlers, enhance error handling, and simplify tree creation logic.

Author: ihongda

sutta_publisher/src/sutta_publisher/edition_parsers/base.py

@@ -9,7 +9,7 @@
 from typing import Callable, cast
 
 import jinja2
-import requests
+import requests  # type: ignore[import-untyped]
 from bs4 import BeautifulSoup, Tag
 from jinja2 import Environment, FileSystemLoader, Template, TemplateNotFound
 
@@ -615,7 +615,7 @@ def _process_html_matter(matter: str, working_dir: str) -> str:
         else:
             response = requests.get(FRONTMATTER_URL.format(matter=matter, working_dir=working_dir))
             response.raise_for_status()
-            return response.text
+            return cast(str, response.text)
 
     @staticmethod
     def _get_template(name: str) -> Template:

sutta_publisher/src/sutta_publisher/edition_parsers/helper_functions.py

@@ -7,7 +7,7 @@
 from typing import Any, cast, no_type_check
 from zipfile import ZipFile
 
-import requests
+import requests  # type: ignore[import-untyped]
 from bs4 import BeautifulSoup, Tag
 from ebooklib.epub import Link, Section
 
@@ -322,7 +322,7 @@ def get_true_volume_index(volume: Volume) -> int:
         return cast(int, volume.volume_number - 1)
 
 
-def get_individual_cover_template_name(publication_type, volume: Volume) -> str:
+def get_individual_cover_template_name(publication_type: str, volume: Volume) -> str:
     """Get volume's individual cover template name"""
     if volume.volume_number and publication_type == "hardcover":
         return f"{volume.text_uid}-{volume.volume_number}-hardcover.tex"
@@ -393,7 +393,4 @@ def make_hardcover_zip_files(paths: list[Path], num_of_volumes: int) -> list[Pat
         (f"{base_filename}-hardcover-cover.zip", cover_files),
         (f"{base_filename}-hardcover.zip", content_files),
     )
-    return [
-        _make_zip(filename=_filename, paths=_paths)
-        for _filename, _paths in mapping
-    ]
+    return [_make_zip(filename=_filename, paths=_paths) for _filename, _paths in mapping]

sutta_publisher/src/sutta_publisher/edition_parsers/latex.py

@@ -286,7 +286,7 @@ def _append_section_title(self, tag: Tag) -> str:
             if self.config.edition.text_uid == "sn":
                 _heading_depth -= 1
                 # if not _heading_depth:
-                    # return ""
+                # return ""
 
             if self.sutta_depth == 2:
                 index = _heading_depth
@@ -557,7 +557,12 @@ def _get_individual_template(self, volume: Volume) -> Template | None:
             log.warning(f"Template '{_template_name}' for publication/volume specific configuration not found.")
             return None
 
-    def _get_cover_template(self, name: str, finalize: Callable[[Any], str] = None, template_dir: str = "") -> Template:
+    def _get_cover_template(
+        self,
+        name: str,
+        finalize: Callable[[Any], str] | None = None,
+        template_dir: str = "",
+    ) -> Template:
 
         _subdir = LatexParser.COVER_TEMPLATES_SUBDIR
 

sutta_publisher/src/sutta_publisher/shared/config.py

@@ -2,7 +2,7 @@
 
 import logging
 
-import requests
+import requests  # type: ignore[import-untyped]
 from pydantic import ValidationError
 
 from sutta_publisher.shared import API_ENDPOINTS, API_URL, CREATOR_BIOS_URL

sutta_publisher/src/sutta_publisher/shared/data.py

@@ -2,7 +2,7 @@
 
 from typing import NoReturn
 
-import requests
+import requests  # type: ignore[import-untyped]
 
 from sutta_publisher.shared import API_ENDPOINTS, API_URL, SUPER_TREE_URL, TREE_URL
 from sutta_publisher.shared.value_objects.edition_config import EditionConfig, Volumes

sutta_publisher/src/sutta_publisher/shared/edition_finder.py

@@ -1,4 +1,4 @@
-from requests import Response
+from requests import Response  # type: ignore[import-untyped]
 
 from sutta_publisher.shared import EDITION_FINDER_PATTERNS, LAST_RUN_DATE_FILE_URL, SCDATA_REPO_URL, SUPER_TREE_URL
 from sutta_publisher.shared.github_handler import get_last_commit_sha, get_modified_filenames, worker

sutta_publisher/src/sutta_publisher/shared/github_handler.py

@@ -6,9 +6,10 @@
 from datetime import datetime
 from pathlib import Path
 from time import sleep
+from typing import Optional
 
-import requests
-from requests import Response
+import requests  # type: ignore[import-untyped]
+from requests import Response  # type: ignore[import-untyped]
 
 from sutta_publisher.shared import EDITIONS_REPO_URL, LAST_RUN_DATE_FILE_URL
 from sutta_publisher.shared.value_objects.edition import EditionResult
@@ -19,7 +20,7 @@
 ERROR_SLEEP_TIME = 1  # in seconds
 
 
-def worker(queue: list[dict], api_key: str = None, silent: bool = False) -> list[Response]:
+def worker(queue: list[dict], api_key: Optional[str] = None, silent: bool = False) -> list[Response]:
 
     _queue: list[tuple[int, dict]] = [(_i, _t) for _i, _t in enumerate(queue)]
     errors = 0
@@ -62,15 +63,37 @@ def worker(queue: list[dict], api_key: str = None, silent: bool = False) -> list
 
 def get_last_commit_sha(repo_url: str, api_key: str, branch: str) -> str:
     """Get SHA of the last commit"""
+    last_commit_sha, _ = get_last_commit_and_tree_sha(repo_url=repo_url, api_key=api_key, branch=branch)
+    return last_commit_sha
+
+
+def get_last_commit_and_tree_sha(repo_url: str, api_key: str, branch: str) -> tuple[str, str]:
+    """Get SHAs of the last commit and its tree."""
     _request = {
         "method": "get",
         "url": f"{repo_url}/branches/{branch}",
         "help_text": "get last commit sha",
     }
     _response: Response = worker(queue=[_request], api_key=api_key)[0]
 
-    sha: str = _response.json()["commit"]["sha"]
-    return sha
+    _response_data = _response.json()
+    _commit = _response_data.get("commit")
+    if not isinstance(_commit, dict):
+        raise ValueError(f"Unexpected GitHub response for branch '{branch}': missing 'commit' object.")
+
+    commit_sha = _commit.get("sha")
+    _nested_commit = _commit.get("commit")
+    if not isinstance(_nested_commit, dict):
+        raise ValueError(f"Unexpected GitHub response for branch '{branch}': missing nested 'commit' object.")
+
+    _tree = _nested_commit.get("tree")
+    if not isinstance(_tree, dict):
+        raise ValueError(f"Unexpected GitHub response for branch '{branch}': missing 'tree' object.")
+
+    tree_sha = _tree.get("sha")
+    if not commit_sha or not tree_sha:
+        raise ValueError(f"Unexpected GitHub response for branch '{branch}': missing commit SHA or tree SHA.")
+    return commit_sha, tree_sha
 
 
 def get_blob_shas(file_paths: list[Path], repo_url: str, api_key: str) -> list[str]:
@@ -129,46 +152,27 @@ def get_old_files_shas(file_paths: list[Path], repo_url: str, repo_path: str, ap
 
 def create_new_tree(
     file_paths: list[Path],
-    repo_url: str,
     repo_path: str,
-    api_key: str,
-    last_commit_sha: str,
     blob_shas: list[str],
-    old_files_shas: list[str],
 ) -> list[dict]:
-    """Create new Git tree with updated files"""
-    _request = {
-        "method": "get",
-        "url": f"{repo_url}/git/trees/{last_commit_sha}?recursive=1",
-        "help_text": "create new tree",
-    }
-    _responses: list[Response] = worker(queue=[_request], api_key=api_key, silent=True)
-
-    _old_tree = _responses[0].json().get("tree", []) if _responses else []
-
-    new_tree: list[dict] = [
-        _item for _item in _old_tree if _item.get("type") == "blob" and not _item.get("sha") in old_files_shas
+    """Create incremental Git tree entries with updated files only."""
+    return [
+        {
+            "path": f"{repo_path}{'/' if repo_path else ''}{_file.name}",
+            "mode": "100644",
+            "type": "blob",
+            "sha": _sha,
+        }
+        for _file, _sha in zip(file_paths, blob_shas)
     ]
-    new_tree.extend(
-        [
-            {
-                "path": f"{repo_path}{'/' if repo_path else ''}{_file.name}",
-                "mode": "100644",
-                "type": "blob",
-                "sha": _sha,
-            }
-            for _file, _sha in zip(file_paths, blob_shas)
-        ]
-    )
-    return new_tree
 
 
-def get_tree_sha(repo_url: str, api_key: str, tree: list[dict]) -> str:
-    """Post a new tree and return its SHA"""
+def get_tree_sha(repo_url: str, api_key: str, tree: list[dict], base_tree_sha: str) -> str:
+    """Post a new tree on top of base tree and return its SHA."""
     _request = {
         "method": "post",
         "url": f"{repo_url}/git/trees",
-        "body": json.dumps({"tree": tree}),
+        "body": json.dumps({"base_tree": base_tree_sha, "tree": tree}),
         "help_text": "create new tree",
     }
     _response: Response = worker(queue=[_request], api_key=api_key)[0]
@@ -217,20 +221,16 @@ def update_head(repo_url: str, api_key: str, new_commit_sha: str) -> None:
 
 
 def upload_files_to_repo(
-    file_paths: list[Path], repo_url: str, repo_path: str, api_key: str, edition: EditionResult = None
+    file_paths: list[Path], repo_url: str, repo_path: str, api_key: str, edition: Optional[EditionResult] = None
 ) -> None:
 
-    last_commit_sha: str = get_last_commit_sha(repo_url, api_key, "main")
+    last_commit_sha, base_tree_sha = get_last_commit_and_tree_sha(repo_url=repo_url, api_key=api_key, branch="main")
 
     blob_shas: list[str] = get_blob_shas(file_paths, repo_url, api_key)
 
-    old_files_shas: list[str] = get_old_files_shas(file_paths, repo_url, repo_path, api_key)
-
-    new_tree: list[dict] = create_new_tree(
-        file_paths, repo_url, repo_path, api_key, last_commit_sha, blob_shas, old_files_shas
-    )
+    new_tree: list[dict] = create_new_tree(file_paths=file_paths, repo_path=repo_path, blob_shas=blob_shas)
 
-    tree_sha: str = get_tree_sha(repo_url, api_key, new_tree)
+    tree_sha: str = get_tree_sha(repo_url=repo_url, api_key=api_key, tree=new_tree, base_tree_sha=base_tree_sha)
 
     new_commit_sha: str = get_new_commit_sha(edition, file_paths, repo_url, api_key, last_commit_sha, tree_sha)