diff --git a/homely/_utils.py b/homely/_utils.py index a9a0f83..3536a46 100644 --- a/homely/_utils.py +++ b/homely/_utils.py @@ -268,10 +268,10 @@ def writejson(self) -> None: class RepoListEntry(TypedDict): repoid: str - localrepo: str + localrepo: dict[str, str | bool] localpath: str canonicalpath: NotRequired[str] - canonicalrepo: NotRequired[str] + canonicalrepo: NotRequired[dict[str, str | bool]] class RepoListConfig(JsonConfig[list[RepoListEntry]]): diff --git a/homely/_vcs/__init__.py b/homely/_vcs/__init__.py index 26c8df5..8634d78 100644 --- a/homely/_vcs/__init__.py +++ b/homely/_vcs/__init__.py @@ -1,14 +1,18 @@ import os +from enum import Enum +from typing import Iterable, Optional from homely._errors import NotARepo _handlers = None -HANDLER_GIT_v1 = "vcs:git" -HANDLER_TESTHANDLER_v1 = "vcs:testhandler" +class RepoType(Enum): + HANDLER_GIT_v1 = "vcs:git" + HANDLER_TESTHANDLER_v1 = "vcs:testhandler" -def _gethandlers(): + +def _gethandlers() -> Iterable[type["Repo"]]: global _handlers import homely._vcs.git import homely._vcs.testhandler @@ -19,7 +23,7 @@ def _gethandlers(): return _handlers -def getrepohandler(repo_path): +def getrepohandler(repo_path: str) -> "Repo": for class_ in _handlers or _gethandlers(): repo = class_.frompath(repo_path) if repo is not None: @@ -27,7 +31,7 @@ def getrepohandler(repo_path): raise NotARepo(repo_path) -def fromdict(row): +def fromdict(row: dict[str, str | bool]) -> "Repo": for class_ in _handlers or _gethandlers(): obj = class_.fromdict(row) if obj is not None: @@ -39,12 +43,16 @@ class Repo: """ Base class for VCS handlers """ - def __init__(self, - repo_path, - isremote, - iscanonical, - suggestedlocal, - canonical=None): + type_: RepoType + + def __init__( + self, + repo_path: str, + isremote: bool, + iscanonical: bool, + suggestedlocal: Optional[str], + canonical: Optional[str] = None, + ): self.isremote = isremote self.iscanonical = iscanonical self.repo_path = repo_path @@ -52,22 +60,20 @@ def __init__(self, suggestedlocal = os.path.basename(repo_path) self.suggestedlocal = suggestedlocal self._canonical = canonical - super(Repo, self).__init__() - assert self.type in (HANDLER_GIT_v1, HANDLER_TESTHANDLER_v1) @classmethod - def frompath(class_, repo_path): + def frompath(class_, repo_path: str) -> Optional["Repo"]: raise Exception( "%s.%s needs to implement @classmethod .frompath(repo_path)" % ( class_.__module__, class_.__name__)) @classmethod - def shortid(class_, repoid): + def shortid(class_, repoid: str) -> str: raise Exception( "%s.%s needs to implement @staticmethod .shortid(repoid)" % ( class_.__module__, class_.__name__)) - def getrepoid(self): + def getrepoid(self) -> str: """ Get a unique id for the repo. For example, the first commit hash. """ @@ -75,7 +81,7 @@ def getrepoid(self): "%s.%s needs to implement .getrepoid()" % ( self.__class__.__module__, self.__class__.__name__)) - def clonetopath(self, dest): + def clonetopath(self, dest: str) -> None: """ Clone the repo at into Note that if self.pushablepath is None, then self.path will be used @@ -85,19 +91,19 @@ def clonetopath(self, dest): "%s.%s needs to implement @classmethod .clonetopath(dest)" % ( self.__class__.__module__, self.__class__.__name__)) - def isdirty(self): + def isdirty(self) -> bool: raise Exception( "%s.%s needs to implement .isdirty()" % ( self.__class__.__module__, self.__class__.__name__)) - def pullchanges(self): + def pullchanges(self) -> None: raise Exception( "%s.%s needs to implement .pullchanges()" % ( self.__class__.__module__, self.__class__.__name__)) - def asdict(self): + def asdict(self) -> dict[str, str | bool]: return dict( - type=self.type, + type=self.type_.value, repo_path=self.repo_path, isremote=self.isremote, iscanonical=self.iscanonical, @@ -105,10 +111,18 @@ def asdict(self): ) @classmethod - def fromdict(class_, row): - if row["type"] == class_.type: - return class_(row["repo_path"], - row["isremote"], - row["iscanonical"], - row["suggestedlocal"], - ) + def fromdict(class_, row: dict[str, str | bool]) -> Optional["Repo"]: + if row["type"] != class_.type_.value: + return None + + assert isinstance(row["repo_path"], str) + assert isinstance(row["suggestedlocal"], str) + assert isinstance(row["isremote"], bool) + assert isinstance(row["iscanonical"], bool) + + return class_( + row["repo_path"], + row["isremote"], + row["iscanonical"], + row["suggestedlocal"], + ) diff --git a/homely/_vcs/git.py b/homely/_vcs/git.py index f149923..382342b 100644 --- a/homely/_vcs/git.py +++ b/homely/_vcs/git.py @@ -1,5 +1,6 @@ import os import re +from typing import Optional import homely._vcs from homely._errors import ConnectionError, RepoError, RepoHasNoCommitsError @@ -8,11 +9,11 @@ class Repo(homely._vcs.Repo): - type = homely._vcs.HANDLER_GIT_v1 + type_ = homely._vcs.RepoType.HANDLER_GIT_v1 pulldesc = 'git pull' @classmethod - def _from_parts(class_, repo_path, user, domain, name): + def _from_parts(class_, repo_path: str, user: str, domain: str, name: str) -> "Repo": if name.endswith('.git'): name = name[0:-4] @@ -27,15 +28,17 @@ def _from_parts(class_, repo_path, user, domain, name): ) @classmethod - def frompath(class_, repo_path): + def frompath(class_, repo_path: str) -> Optional["Repo"]: if os.path.isdir(repo_path): if not os.path.isdir(os.path.join(repo_path, '.git')): - return + return None + return class_(_expandpath(repo_path), isremote=False, iscanonical=False, suggestedlocal=None ) + if (repo_path.startswith('ssh://') or repo_path.startswith('https://') or repo_path.startswith('git@')): @@ -58,7 +61,9 @@ def frompath(class_, repo_path): iscanonical=False, suggestedlocal=None) - def pullchanges(self): + return None + + def pullchanges(self) -> None: assert not self.isremote cmd = ['git', 'pull'] code, _, err = execute(cmd, @@ -69,23 +74,25 @@ def pullchanges(self): return assert code == 1 + assert isinstance(err, bytes) # TODO: replace this with a type check needle = b'fatal: Could not read from remote repository.' if needle in err: raise ConnectionError() - raise SystemError("Unexpected output from 'git pull': {}".format(err)) + raise SystemError(f"Unexpected output from 'git pull': {err!r}") - def clonetopath(self, dest): + def clonetopath(self, dest: str) -> None: origin = self.repo_path execute(['git', 'clone', origin, dest]) - def getrepoid(self): + def getrepoid(self) -> str: assert not self.isremote cmd = ['git', 'rev-list', '--max-parents=0', 'HEAD'] returncode, stdout = run(cmd, cwd=self.repo_path, stdout=True, stderr="STDOUT")[:2] + assert isinstance(stdout, bytes) # TODO: replace this with a type check if returncode == 0: return self._getfirsthash(stdout) if returncode != 128: @@ -115,19 +122,20 @@ def getrepoid(self): raise SystemError("Unexpected exitcode {}".format(returncode)) - def _getfirsthash(self, stdout): + def _getfirsthash(self, stdout: bytes) -> str: stripped = stdout.rstrip().decode('utf-8') if '\n' in stripped: raise RepoError("Git repo has multiple initial commits") return stripped @staticmethod - def shortid(repoid): + def shortid(repoid: str) -> str: return repoid[0:8] - def isdirty(self): + def isdirty(self) -> bool: cmd = ['git', 'status', '--porcelain'] out = execute(cmd, cwd=self.repo_path, stdout=True)[1] + assert isinstance(out, bytes) # TODO: replace this with a type check for line in out.split(b'\n'): if len(line) and not line.startswith(b'?? '): return True diff --git a/homely/_vcs/testhandler.py b/homely/_vcs/testhandler.py index 239c107..c327d77 100644 --- a/homely/_vcs/testhandler.py +++ b/homely/_vcs/testhandler.py @@ -13,11 +13,11 @@ class Repo(homely._vcs.Repo): - type = homely._vcs.HANDLER_TESTHANDLER_v1 + type_ = homely._vcs.RepoType.HANDLER_TESTHANDLER_v1 pulldesc = 'fake repo pull' @classmethod - def frompath(class_, repo_path) -> "Optional[homely._vcs.Repo]": + def frompath(class_, repo_path: str) -> "Optional[homely._vcs.Repo]": if repo_path.startswith(PREFIX): dirpart = repo_path[len(PREFIX):] return class_(repo_path, @@ -37,7 +37,7 @@ def frompath(class_, repo_path) -> "Optional[homely._vcs.Repo]": iscanonical=False, suggestedlocal=None) - def clonetopath(self, dest_path): + def clonetopath(self, dest_path: str) -> None: assert not os.path.exists(dest_path) os.mkdir(dest_path) with open(os.path.join(dest_path, ORIGINFILE), 'w') as f: @@ -47,7 +47,7 @@ def clonetopath(self, dest_path): origin = origin[len(PREFIX):] self._pull(origin, dest_path) - def _pull(self, origin, local): + def _pull(self, origin: str, local: str) -> None: # delete every local file except the special ones for thing in os.listdir(local): if thing not in (ORIGINFILE, MARKERFILE, DIRTYFILE): @@ -70,7 +70,7 @@ def _pull(self, origin, local): else: shutil.copy2(src, dst) - def pullchanges(self): + def pullchanges(self) -> None: assert not self.isdirty() with open(os.path.join(self.repo_path, ORIGINFILE), 'r') as f: origin = f.read().strip() @@ -78,15 +78,15 @@ def pullchanges(self): origin = origin[len(PREFIX):] self._pull(origin, self.repo_path) - def getrepoid(self): + def getrepoid(self) -> str: assert not self.isremote with open(os.path.join(self.repo_path, MARKERFILE), 'r') as f: return f.read().strip() @staticmethod - def shortid(repoid): + def shortid(repoid: str) -> str: return repoid[0:5] - def isdirty(self): + def isdirty(self) -> bool: assert not self.isremote return os.path.exists(os.path.join(self.repo_path, DIRTYFILE)) diff --git a/pyproject.toml b/pyproject.toml index fb586da..cfbd83b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,9 +70,6 @@ module = [ "homely._test.system", "homely._test", "homely._ui", - "homely._vcs.git", - "homely._vcs.testhandler", - "homely._vcs", "homely.files", "homely.general", "homely.install",