Merge pull request #385 from legendu-net/dev

Merge dev into main

Author: dclong

aiutil/__init__.py

@@ -2,4 +2,4 @@
 """
 from . import poetry
 
-__version__ = "0.80.6"
+__version__ = "0.81.0"

aiutil/git.py

@@ -0,0 +1,71 @@
+"""A module providing wrap over the git command.
+"""
+from pathlib import Path
+import subprocess as sp
+
+
+class Repo:
+    """A local Git repository."""
+
+    def __init__(self, root: str | Path):
+        self._root = str(root)
+        self._remote = self.remote()
+
+    def remote(self) -> str:
+        """Run the "git remote" command."""
+        proc = sp.run(
+            ["git", "-C", self._root, "remote"], check=True, capture_output=True
+        )
+        return proc.stdout.strip().decode()
+
+    def pull(self, branch: str) -> None:
+        """Run the "git pull" command."""
+        sp.run(["git", "-C", self._root, "pull", self._remote, branch], check=True)
+
+    def push(self, branch: str = "") -> None:
+        """Run the "git push" command."""
+        if not branch:
+            branch = self.active_branch()
+        sp.run(["git", "-C", self._root, "push", self._remote, branch], check=True)
+
+    def _branch(self) -> list[str]:
+        proc = sp.run(
+            ["git", "-C", self._root, "branch"], check=True, capture_output=True
+        )
+        return proc.stdout.strip().decode().split("\n")
+
+    def active_branch(self) -> str:
+        """Get the active branch of the repository."""
+        lines = self._branch()
+        return next(line[2:] for line in lines if line.startswith("* "))
+
+    def branch(self):
+        """Run the "git branch" command to get all branches of the repository."""
+        return [b[2:] for b in self._branch()]
+
+    def add(self, pattern: str = "."):
+        """Run the "git add" command."""
+        sp.run(["git", "-C", self._root, "add", pattern], check=True)
+
+    def commit(self, message: str):
+        """Run the "git commit" command."""
+        sp.run(["git", "-C", self._root, "commit", "-m", message], check=True)
+
+    def status(self):
+        """Run the "git status" command."""
+        sp.run(["git", "-C", self._root, "status"], check=True)
+
+    def tag(self, tag="") -> sp.CompletedProcess | list[str]:
+        """Run the "git tag" command."""
+        if tag:
+            return sp.run(["git", "tag", tag], check=True)
+        proc = sp.run(["git", "tag"], check=True, capture_output=True)
+        return proc.stdout.strip().decode().split("\n")
+
+    def checkout(self, branch: str = "", new: str = ""):
+        """Run the "git checkout" command."""
+        if new:
+            return sp.run(["git", "checkout", "-b", new], check=True)
+        if not branch:
+            raise ValueError("Either new or branch is required!")
+        return sp.run(["git", "checkout", branch], check=True)

aiutil/poetry.py

@@ -9,14 +9,12 @@
 import toml
 from loguru import logger
 import pathspec
-import dulwich
-from dulwich.contrib.paramiko_vendor import ParamikoSSHVendor
 from .filesystem import replace_patterns
+from . import git
 
 DIST = "dist"
 README = "README.md"
 TOML = "pyproject.toml"
-dulwich.client.get_ssh_vendor = ParamikoSSHVendor
 
 
 def _project_dir() -> Path:
@@ -30,9 +28,7 @@ def _project_dir() -> Path:
         if (path / TOML).is_file():
             return path
         path = path.parent
-    raise RuntimeError(
-        f"The current work directory {Path.cwd()} is not a (subdirectory of a) Python Poetry project."
-    )
+    return Path()
 
 
 def _project_name(proj_dir: Path) -> str:
@@ -118,10 +114,10 @@ def version(
     if ver:
         _update_version(ver=ver, proj_dir=proj_dir)
         if commit:
-            repo = dulwich.repo.Repo(proj_dir)
-            dulwich.porcelain.add(repo=repo)
-            dulwich.porcelain.commit(repo=repo, message="bump up version")
-            dulwich.porcelain.push(repo=repo)
+            repo = git.Repo(root=proj_dir)
+            repo.add()
+            repo.commit(message="bump up version")
+            repo.push()
     else:
         print(_project_version(proj_dir))
 
@@ -137,25 +133,23 @@ def add_tag_release(
     :param branch_release: The branch for releasing.
     :raises ValueError: If the tag to create already exists.
     """
+    if proj_dir is None:
+        proj_dir = _project_dir()
     if not tag:
-        if proj_dir is None:
-            proj_dir = _project_dir()
         tag = "v" + _project_version(proj_dir)
-    repo = dulwich.repo.Repo(proj_dir)
-    if tag.encode() in dulwich.porcelain.tag_list(repo):
+    repo = git.Repo(proj_dir)
+    if tag.encode() in repo.tag():
         raise ValueError(
             f"The tag {tag} already exists! Please merge new changes to the {branch_release} branch first."
         )
-    branch_old = dulwich.porcelain.active_branch(repo=repo).decode()
+    branch_old = repo.active_branch()
     # add tag to the release branch
-    dulwich.porcelain.checkout_branch(repo=repo, target=branch_release)
-    dulwich.porcelain.pull(repo=repo, refspecs=branch_release)
-    dulwich.porcelain.tag_create(repo=repo, tag=tag, annotated=True)
-    remote = dulwich.porcelain.get_branch_remote(repo=repo).decode()
-    sp.run(f"git push {remote} {tag}", shell=True, check=True)
-    # dulwich.porcelain.push(repo=repo, refspecs=[f"refs/tags/{tag}"])
+    repo.checkout(branch=branch_release)
+    repo.pull(branch=branch_release)
+    repo.tag(tag=tag)
+    repo.push(branch=tag)
     # switch back to the old branch
-    dulwich.porcelain.checkout_branch(repo=repo, target=branch_old)
+    repo.checkout(branch=branch_old)
 
 
 def format_code(
@@ -184,11 +178,11 @@ def format_code(
     logger.info("Formatting code using black ...")
     sp.run(cmd, shell=True, check=False, stdout=sp.PIPE)
     if commit:
-        repo = dulwich.repo.Repo(proj_dir)
-        dulwich.porcelain.add(repo=repo)
-        dulwich.porcelain.commit(repo=repo, message="format code")
-        dulwich.porcelain.push(repo=repo)
-        print(dulwich.porcelain.status())
+        repo = git.Repo(proj_dir)
+        repo.add()
+        repo.commit(message="format code")
+        repo.push()
+        repo.status()
 
 
 def _lint_code(proj_dir: Union[Path, None], linter: Union[str, list[str]]):