PK!҂kkis_git_repo_clean/__init__.pyfrom .meta import version from .check import check, checkSync __all__ = ["check", "checkSync", "version"] PK! jjis_git_repo_clean/check.py# # TODO: find a cross-platform way to rely only on the returncode for untracked # files. Currently it returns stdout/stderr which should be unnecessary. # # ------- # # Imports # # ------- # from textwrap import dedent import asyncio import os import subprocess from .utils import resolveAll, whenTruthy # ---- # # Init # # ---- # runAsync = asyncio.create_subprocess_exec run = subprocess.run headExistsCmd = ["git", "rev-parse", "HEAD"] isAGitRepoCmd = ["git", "rev-parse", "--is-inside-work-tree"] # Any way to remove the double negative? allFilesAreTrackedCmd = [ "git", "ls-files", "--other", "--directory", "--exclude-standard", "--no-empty-directory", ] emptyGitTree = "4b825dc642cb6eb9a060e54bf8d69288fbee4904" # ---- # # Main # # ---- # async def check(cwd=None): cwd = cleanCwd(cwd) isAGitRepo_proc = await runAsync( *isAGitRepoCmd, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, cwd=cwd, ) isAGitRepo_result = await isAGitRepo_proc.wait() if isAGitRepo_result != 0: raise Exception(f"'{cwd}' is not a git repository") headExists_proc = await runAsync( *headExistsCmd, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, cwd=cwd, ) headExists_result = await headExists_proc.wait() headExists = headExists_result == 0 hasNoChangesCmd = getHasNoChangesCmd(headExists) checks = [ runAsync( *hasNoChangesCmd, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, cwd=cwd, ), runAsync( *allFilesAreTrackedCmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, cwd=cwd, ), ] noChanges_proc, allTracked_proc = await resolveAll(checks) results = await resolveAll( [noChanges_proc.wait(), allTracked_proc.communicate()] ) noChanges_result, allTracked_result = results if allTracked_proc.returncode != 0 or noChanges_result != 0: return False stdout, _stderr = allTracked_result return len(stdout.decode("utf-8")) == 0 def checkSync(cwd=None): cwd = cleanCwd(cwd) isAGitRepo = ( run( isAGitRepoCmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cwd=cwd, ).returncode == 0 ) if not isAGitRepo: raise Exception(f"'{cwd} is not a git repository") headExists = ( run( headExistsCmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cwd=cwd, ).returncode == 0 ) hasNoChangesCmd = getHasNoChangesCmd(headExists) allTracked_result = run( allFilesAreTrackedCmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=cwd, ) hasNoChanges_result = run( hasNoChangesCmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, cwd=cwd, ) return ( hasNoChanges_result.returncode == 0 and allTracked_result.returncode == 0 and len(allTracked_result.stdout) == 0 ) # ------- # # Helpers # # ------- # def cleanCwd(cwd): if cwd is None: return os.getcwd() elif not isinstance(cwd, str): raise TypeError( dedent( f"""\ cwd must be an instance of str type(cwd): {type(cwd)} cwd: {cwd} """ ) ) return cwd # # "NoChanges" means no staged nor unstaged changes # def getHasNoChangesCmd(headExists): return [ "git", "diff-index", "--quiet", "--cached", whenTruthy(headExists).return_("HEAD").otherwise(emptyGitTree), "--", ] PK!@Bq q is_git_repo_clean/cli.py# ------- # # Imports # # ------- # from .meta import version from traceback import format_exc from textwrap import dedent from types import SimpleNamespace as o from .check import checkSync from .utils import iif import os # ---- # # Init # # ---- # arguments = set(["--dir", "--silent"]) helpOrVersion = set(["--help", "--version"]) twoLineSeps = os.linesep + os.linesep usage = dedent( f""" Usage is-git-repo-clean [options] is-git-repo-clean (--help | --version) Options --dir: path to the git repo to test. Defaults to `os.getcwd()` --silent: a flag which disables output Returns : 0: yes 1: no 2: 3: dir is not a git repository 4: unexpected error occurred """ ) # ---- # # Main # # ---- # def getIsGitRepoClean(args): result = o(stdout=None, stderr=None, code=None) numArgs = len(args) if numArgs == 1: if args[0] == "--help": result.stdout = usage result.code = 0 return result elif args[0] == "--version": result.stdout = version result.code = 0 return result argsObj = validateAndParseArgs(args, result) if argsObj is result: return result isSilent = argsObj.silent try: isClean = checkSync(argsObj.dir or None) if not isSilent: if isClean: result.stdout = "yes" else: result.stderr = "no" result.code = iif(isClean, 0, 1) return result except Exception as e: if "is not a git repository" in str(e): if not isSilent: result.stderr = "dir is not a git repository" result.code = 3 else: if not isSilent: result.stderr = ( f"unexpected error occurred{twoLineSeps}" + format_exc() ) result.code = 4 return result # ------- # # Helpers # # ------- # def validateAndParseArgs(args, result): i = 0 argsObj = o(dir=None, silent=False) while i < len(args): arg = args[i] if arg not in arguments: if arg in helpOrVersion: result.stderr = f"'{arg}' must be the only argument when passed" else: result.stderr = f"invalid option '{arg}'" result.stderr += os.linesep + usage result.code = 2 return result if arg == "--dir": if i == len(args) - 1: result.stderr = "'--dir' must be given a value" result.stderr += os.linesep + usage result.code = 2 return result argsObj.dir = args[i + 1] i += 1 else: argsObj.silent = True i += 1 return argsObj PK!J^is_git_repo_clean/meta.pyversion = "0.3.0" PK!xCis_git_repo_clean/utils.py# ------- # # Imports # # ------- # from asyncio import gather from types import SimpleNamespace as o # ---- # # Main # # ---- # # # I like to think the common use-case for 'gather' is to cancel all async # operations should any one of them fail. This is not how 'gather' was # implemented though which is why I wrote this function # async def resolveAll(awaitables): gatherAll = None try: gatherAll = gather(*awaitables) result = await gatherAll finally: if gatherAll: gatherAll.cancel() return result # # I have no idea what to name this. Python just doesn't have a clean ternary # operator which stinks and I didn't want to use 'iif' because that's less # readable. So for now we have a verbose and less performant, but readable # function chain # def whenTruthy(condition): def return_(truthyResult): def otherwise(falseyResult): if condition: return truthyResult else: return falseyResult return o(otherwise=otherwise) return o(return_=return_) def iif(condition, whenTruthy, whenFalsey): if condition: return whenTruthy else: return whenFalsey PK!HnHTU'is_git_repo_clean-0.3.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!Hb3bW*is_git_repo_clean-0.3.0.dist-info/METADATAJ1F&TW(7mQaddlֻ3gG*+[2G! E'AЫ_he:RUla%:SCirωD P4,fo6c[<M-SH;pp9P ; 2aYG1B>*'87U~p3k cB#я2uZ}PK!H((is_git_repo_clean-0.3.0.dist-info/RECORD˒0}?Ke1 )EEZ{ D3],csAvS-n!.E,/?@ӫuM]7VJX,*ї`\/N$Q?PTP$X恫l S{i6`ͅTW$Ov}>^7zy\AV0t%66~Nr o]$ڭ%y}zspr@C2]Fx=:LW;J7@vkO)9v~9"$YQ~P"żpϭmT>]VK%םP0rOEc ݵևu8N:5B1I۔ 䫣HF (o}{}?xPK!҂kkis_git_repo_clean/__init__.pyPK! jjis_git_repo_clean/check.pyPK!@Bq q His_git_repo_clean/cli.pyPK!J^is_git_repo_clean/meta.pyPK!xC8is_git_repo_clean/utils.pyPK!HnHTU'A!is_git_repo_clean-0.3.0.dist-info/WHEELPK!Hb3bW*!is_git_repo_clean-0.3.0.dist-info/METADATAPK!H(("is_git_repo_clean-0.3.0.dist-info/RECORDPKk$