git/gita
1
0
Fork 0
Code Issues Activity

Adding upstream version 0.10.9.

Browse source 
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
Daniel Baumann 2025-02-11 18:25:06 +01:00
parent c0f23aff1f
commit a9c588f707
Signed by: daniel
GPG key ID: FBB4F0E80A80222F
27 changed files with 1822 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

3
gita/__init__.py Normal file
View file

@ -0,0 +1,3 @@
import pkg_resources
__version__ = pkg_resources.get_distribution('gita').version

289
gita/__main__.py Normal file
View file

@ -0,0 +1,289 @@
'''
Gita manages multiple git repos. It has two functionalities
1. display the status of multiple repos side by side
2. delegate git commands/aliases from any working directory
Examples:
gita ls
gita fetch
gita stat myrepo2
gita super myrepo1 commit -am 'add some cool feature'
For bash auto completion, download and source
https://github.com/nosarthur/gita/blob/master/.gita-completion.bash
'''
import os
import argparse
import subprocess
import pkg_resources
from . import utils, info
def f_add(args: argparse.Namespace):
repos = utils.get_repos()
utils.add_repos(repos, args.paths)
def f_rename(args: argparse.Namespace):
repos = utils.get_repos()
utils.rename_repo(repos, args.repo[0], args.new_name)
def f_info(_):
all_items, to_display = info.get_info_items()
print('In use:', ','.join(to_display))
unused = set(all_items) - set(to_display)
if unused:
print('Unused:', ' '.join(unused))
def f_ll(args: argparse.Namespace):
"""
Display details of all repos
"""
repos = utils.get_repos()
if args.group: # only display repos in this group
group_repos = utils.get_groups()[args.group]
repos = {k: repos[k] for k in group_repos if k in repos}
for line in utils.describe(repos):
print(line)
def f_ls(args: argparse.Namespace):
repos = utils.get_repos()
if args.repo: # one repo, show its path
print(repos[args.repo])
else: # show names of all repos
print(' '.join(repos))
def f_group(args: argparse.Namespace):
repos = utils.get_repos()
groups = utils.get_groups()
if args.to_group:
gname = input('group name? ')
if gname in groups:
gname_repos = set(groups[gname])
gname_repos.update(args.to_group)
groups[gname] = sorted(gname_repos)
utils.write_to_groups_file(groups, 'w')
else:
utils.write_to_groups_file({gname: sorted(args.to_group)}, 'a+')
else:
for group, repos in groups.items():
print(f"{group}: {', '.join(repos)}")
def f_ungroup(args: argparse.Namespace):
groups = utils.get_groups()
to_ungroup = set(args.to_ungroup)
to_del = []
for name, repos in groups.items():
remaining = set(repos) - to_ungroup
if remaining:
groups[name] = list(sorted(remaining))
else:
to_del.append(name)
for name in to_del:
del groups[name]
utils.write_to_groups_file(groups, 'w')
def f_rm(args: argparse.Namespace):
"""
Unregister repo(s) from gita
"""
path_file = utils.get_config_fname('repo_path')
if os.path.isfile(path_file):
repos = utils.get_repos()
for repo in args.repo:
del repos[repo]
utils.write_to_repo_file(repos, 'w')
def f_git_cmd(args: argparse.Namespace):
"""
Delegate git command/alias defined in `args.cmd`. Asynchronous execution is
disabled for commands in the `args.async_blacklist`.
"""
repos = utils.get_repos()
groups = utils.get_groups()
if args.repo: # with user specified repo(s) or group(s)
chosen = {}
for k in args.repo:
if k in repos:
chosen[k] = repos[k]
if k in groups:
for r in groups[k]:
chosen[r] = repos[r]
repos = chosen
cmds = ['git'] + args.cmd
if len(repos) == 1 or cmds[1] in args.async_blacklist:
for path in repos.values():
print(path)
subprocess.run(cmds, cwd=path)
else: # run concurrent subprocesses
# Async execution cannot deal with multiple repos' user name/password.
# Here we shut off any user input in the async execution, and re-run
# the failed ones synchronously.
errors = utils.exec_async_tasks(
utils.run_async(repo_name, path, cmds) for repo_name, path in repos.items())
for path in errors:
if path:
print(path)
subprocess.run(cmds, cwd=path)
def f_super(args):
"""
Delegate git command/alias defined in `args.man`, which may or may not
contain repo names.
"""
names = []
repos = utils.get_repos()
groups = utils.get_groups()
for i, word in enumerate(args.man):
if word in repos or word in groups:
names.append(word)
else:
break
args.cmd = args.man[i:]
args.repo = names
f_git_cmd(args)
def main(argv=None):
p = argparse.ArgumentParser(prog='gita',
formatter_class=argparse.RawTextHelpFormatter,
description=__doc__)
subparsers = p.add_subparsers(title='sub-commands',
help='additional help with sub-command -h')
version = pkg_resources.require('gita')[0].version
p.add_argument('-v',
'--version',
action='version',
version=f'%(prog)s {version}')
# bookkeeping sub-commands
p_add = subparsers.add_parser('add', help='add repo(s)')
p_add.add_argument('paths', nargs='+', help="add repo(s)")
p_add.set_defaults(func=f_add)
p_rm = subparsers.add_parser('rm', help='remove repo(s)')
p_rm.add_argument('repo',
nargs='+',
choices=utils.get_repos(),
help="remove the chosen repo(s)")
p_rm.set_defaults(func=f_rm)
p_rename = subparsers.add_parser('rename', help='rename a repo')
p_rename.add_argument(
'repo',
nargs=1,
choices=utils.get_repos(),
help="rename the chosen repo")
p_rename.add_argument(
'new_name',
help="new name")
p_rename.set_defaults(func=f_rename)
p_info = subparsers.add_parser('info', help='show information items of the ll sub-command')
p_info.set_defaults(func=f_info)
ll_doc = f''' status symbols:
+: staged changes
*: unstaged changes
_: untracked files/folders
branch colors:
{info.Color.white}white{info.Color.end}: local has no remote
{info.Color.green}green{info.Color.end}: local is the same as remote
{info.Color.red}red{info.Color.end}: local has diverged from remote
{info.Color.purple}purple{info.Color.end}: local is ahead of remote (good for push)
{info.Color.yellow}yellow{info.Color.end}: local is behind remote (good for merge)'''
p_ll = subparsers.add_parser('ll',
help='display summary of all repos',
formatter_class=argparse.RawTextHelpFormatter,
description=ll_doc)
p_ll.add_argument('group',
nargs='?',
choices=utils.get_groups(),
help="show repos in the chosen group")
p_ll.set_defaults(func=f_ll)
p_ls = subparsers.add_parser(
'ls', help='display names of all repos, or path of a chosen repo')
p_ls.add_argument('repo',
nargs='?',
choices=utils.get_repos(),
help="show path of the chosen repo")
p_ls.set_defaults(func=f_ls)
p_group = subparsers.add_parser(
'group', help='group repos or display names of all groups if no repo is provided')
p_group.add_argument('to_group',
nargs='*',
choices=utils.get_choices(),
help="repo(s) to be grouped")
p_group.set_defaults(func=f_group)
p_ungroup = subparsers.add_parser(
'ungroup', help='remove group information for repos',
description="Remove group information on repos")
p_ungroup.add_argument('to_ungroup',
nargs='+',
choices=utils.get_repos(),
help="repo(s) to be ungrouped")
p_ungroup.set_defaults(func=f_ungroup)
# superman mode
p_super = subparsers.add_parser(
'super',
help='superman mode: delegate any git command/alias in specified or '
'all repo(s).\n'
'Examples:\n \t gita super myrepo1 commit -am "fix a bug"\n'
'\t gita super repo1 repo2 repo3 checkout new-feature')
p_super.add_argument(
'man',
nargs=argparse.REMAINDER,
help="execute arbitrary git command/alias for specified or all repos "
"Example: gita super myrepo1 diff --name-only --staged "
"Another: gita super checkout master ")
p_super.set_defaults(func=f_super)
# sub-commands that fit boilerplate
cmds = utils.get_cmds_from_files()
for name, data in cmds.items():
help = data.get('help')
cmd = data.get('cmd') or name
if data.get('allow_all'):
choices = utils.get_choices()
nargs = '*'
help += ' for all repos or'
else:
choices = utils.get_repos().keys() | utils.get_groups().keys()
nargs = '+'
help += ' for the chosen repo(s) or group(s)'
sp = subparsers.add_parser(name, help=help)
sp.add_argument('repo', nargs=nargs, choices=choices, help=help)
sp.set_defaults(func=f_git_cmd, cmd=cmd.split())
args = p.parse_args(argv)
args.async_blacklist = {
name
for name, data in cmds.items() if data.get('disable_async')
}
if 'func' in args:
args.func(args)
else:
p.print_help() # pragma: no cover
if __name__ == '__main__':
main() # pragma: no cover

65
gita/cmds.yml Normal file
View file

@ -0,0 +1,65 @@
br:
cmd: branch -vv
help: show local branches
clean:
cmd: clean -dfx
help: remove all untracked files/folders
diff:
help: show differences
difftool:
disable_async: true
help: show differences using a tool
fetch:
allow_all: true
help: fetch remote update
last:
cmd: log -1 HEAD
help: show log information of HEAD
log:
disable_async: true
help: show logs
merge:
cmd: merge @{u}
help: merge remote updates
mergetool:
disable_async: true
help: merge updates with a tool
patch:
cmd: format-patch HEAD~
help: make a patch
pull:
allow_all: true
help: pull remote updates
push:
help: push the local updates
rebase:
help: rebase from master
reflog:
help: show ref logs
remote:
cmd: remote -v
help: show remote settings
reset:
help: reset repo(s)
shortlog:
disable_async: true
help: show short log
show:
disable_async: true
help: show detailed commit information
show-branch:
disable_async: true
help: show detailed branch information
stash:
help: store uncommited changes
stat:
cmd: diff --stat
help: show edit statistics
st:
help: show status
tag:
cmd: tag -n
help: show tags
whatchanged:
disable_async: true
help: show detailed log

7
gita/common.py Normal file
View file

@ -0,0 +1,7 @@
import os
def get_config_dir() -> str:
parent = os.environ.get('XDG_CONFIG_HOME') or os.path.join(
os.path.expanduser('~'), '.config')
root = os.path.join(parent,"gita")
return root

146
gita/info.py Normal file
View file

@ -0,0 +1,146 @@
import os
import sys
import yaml
import subprocess
from typing import Tuple, List, Callable, Dict
from . import common
class Color:
"""
Terminal color
"""
red = '\x1b[31m' # local diverges from remote
green = '\x1b[32m' # local == remote
yellow = '\x1b[33m' # local is behind
blue = '\x1b[34m'
purple = '\x1b[35m' # local is ahead
cyan = '\x1b[36m'
white = '\x1b[37m' # no remote branch
end = '\x1b[0m'
def get_info_funcs() -> List[Callable[[str], str]]:
"""
Return the functions to generate `gita ll` information. All these functions
take the repo path as input and return the corresponding information as str.
See `get_path`, `get_repo_status`, `get_common_commit` for examples.
"""
info_items, to_display = get_info_items()
return [info_items[k] for k in to_display]
def get_info_items() -> Tuple[Dict[str, Callable[[str], str]], List[str]]:
"""
Return the available information items for display in the `gita ll`
sub-command, and the ones to be displayed.
It loads custom information functions and configuration if they exist.
"""
# default settings
info_items = {'branch': get_repo_status,
'commit_msg': get_commit_msg,
'path': get_path, }
display_items = ['branch', 'commit_msg']
# custom settings
root = common.get_config_dir()
src_fname = os.path.join(root, 'extra_repo_info.py')
yml_fname = os.path.join(root, 'info.yml')
if os.path.isfile(src_fname):
sys.path.append(root)
from extra_repo_info import extra_info_items
info_items.update(extra_info_items)
if os.path.isfile(yml_fname):
with open(yml_fname, 'r') as stream:
display_items = yaml.load(stream, Loader=yaml.FullLoader)
display_items = [x for x in display_items if x in info_items]
return info_items, display_items
def get_path(path):
return Color.cyan + path + Color.end
def get_head(path: str) -> str:
result = subprocess.run('git rev-parse --abbrev-ref HEAD'.split(),
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
universal_newlines=True,
cwd=path)
return result.stdout.strip()
def run_quiet_diff(args: List[str]) -> bool:
"""
Return the return code of git diff `args` in quiet mode
"""
result = subprocess.run(
['git', 'diff', '--quiet'] + args,
stderr=subprocess.DEVNULL,
)
return result.returncode
def get_common_commit() -> str:
"""
Return the hash of the common commit of the local and upstream branches.
"""
result = subprocess.run('git merge-base @{0} @{u}'.split(),
stdout=subprocess.PIPE,
universal_newlines=True)
return result.stdout.strip()
def has_untracked() -> bool:
"""
Return True if untracked file/folder exists
"""
result = subprocess.run('git ls-files -zo --exclude-standard'.split(),
stdout=subprocess.PIPE)
return bool(result.stdout)
def get_commit_msg(path: str) -> str:
"""
Return the last commit message.
"""
# `git show-branch --no-name HEAD` is faster than `git show -s --format=%s`
result = subprocess.run('git show-branch --no-name HEAD'.split(),
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
universal_newlines=True,
cwd=path)
return result.stdout.strip()
def get_repo_status(path: str) -> str:
head = get_head(path)
dirty, staged, untracked, color = _get_repo_status(path)
return f'{color}{head+" "+dirty+staged+untracked:<10}{Color.end}'
def _get_repo_status(path: str) -> Tuple[str]:
"""
Return the status of one repo
"""
os.chdir(path)
dirty = '*' if run_quiet_diff([]) else ''
staged = '+' if run_quiet_diff(['--cached']) else ''
untracked = '_' if has_untracked() else ''
diff_returncode = run_quiet_diff(['@{u}', '@{0}'])
has_no_remote = diff_returncode == 128
has_no_diff = diff_returncode == 0
if has_no_remote:
color = Color.white
elif has_no_diff:
color = Color.green
else:
common_commit = get_common_commit()
outdated = run_quiet_diff(['@{u}', common_commit])
if outdated:
diverged = run_quiet_diff(['@{0}', common_commit])
color = Color.red if diverged else Color.yellow
else: # local is ahead of remote
color = Color.purple
return dirty, staged, untracked, color

225
gita/utils.py Normal file
View file

@ -0,0 +1,225 @@
import os
import yaml
import asyncio
import platform
from functools import lru_cache
from typing import List, Dict, Coroutine, Union
from . import info
from . import common
def get_config_fname(fname: str) -> str:
"""
Return the file name that stores the repo locations.
"""
root = common.get_config_dir()
return os.path.join(root, fname)
@lru_cache()
def get_repos() -> Dict[str, str]:
"""
Return a `dict` of repo name to repo absolute path
"""
path_file = get_config_fname('repo_path')
repos = {}
# Each line is a repo path and repo name separated by ,
if os.path.isfile(path_file) and os.stat(path_file).st_size > 0:
with open(path_file) as f:
for line in f:
line = line.rstrip()
if not line: # blank line
continue
path, name = line.split(',')
if not is_git(path):
continue
if name not in repos:
repos[name] = path
else: # repo name collision for different paths: include parent path name
par_name = os.path.basename(os.path.dirname(path))
repos[os.path.join(par_name, name)] = path
return repos
@lru_cache()
def get_groups() -> Dict[str, List[str]]:
"""
Return a `dict` of group name to repo names.
"""
fname = get_config_fname('groups.yml')
groups = {}
# Each line is a repo path and repo name separated by ,
if os.path.isfile(fname) and os.stat(fname).st_size > 0:
with open(fname, 'r') as f:
groups = yaml.load(f, Loader=yaml.FullLoader)
return groups
def get_choices() -> List[Union[str, None]]:
"""
Return all repo names, group names, and an additional empty list. The empty
list is added as a workaround of
argparse's problem with coexisting nargs='*' and choices.
See https://utcc.utoronto.ca/~cks/space/blog/python/ArgparseNargsChoicesLimitation
and
https://bugs.python.org/issue27227
"""
choices = list(get_repos())
choices.extend(get_groups())
choices.append([])
return choices
def is_git(path: str) -> bool:
"""
Return True if the path is a git repo.
"""
# An alternative is to call `git rev-parse --is-inside-work-tree`
# I don't see why that one is better yet.
# For a regular git repo, .git is a folder, for a worktree repo, .git is a file.
# However, git submodule repo also has .git as a file.
# A more reliable way to differentiable regular and worktree repos is to
# compare the result of `git rev-parse --git-dir` and
# `git rev-parse --git-common-dir`
loc = os.path.join(path, '.git')
# TODO: we can display the worktree repos in a different font.
return os.path.exists(loc)
def rename_repo(repos: Dict[str, str], repo: str, new_name: str):
"""
Write new repo name to file
"""
path = repos[repo]
del repos[repo]
repos[new_name] = path
write_to_repo_file(repos, 'w')
def write_to_repo_file(repos: Dict[str, str], mode: str):
"""
"""
data = ''.join(f'{path},{name}\n' for name, path in repos.items())
fname = get_config_fname('repo_path')
os.makedirs(os.path.dirname(fname), exist_ok=True)
with open(fname, mode) as f:
f.write(data)
def write_to_groups_file(groups: Dict[str, List[str]], mode: str):
"""
"""
fname = get_config_fname('groups.yml')
os.makedirs(os.path.dirname(fname), exist_ok=True)
with open(fname, mode) as f:
yaml.dump(groups, f, default_flow_style=None)
def add_repos(repos: Dict[str, str], new_paths: List[str]):
"""
Write new repo paths to file
"""
existing_paths = set(repos.values())
new_paths = set(os.path.abspath(p) for p in new_paths if is_git(p))
new_paths = new_paths - existing_paths
if new_paths:
print(f"Found {len(new_paths)} new repo(s).")
new_repos = {
os.path.basename(os.path.normpath(path)): path
for path in new_paths}
write_to_repo_file(new_repos, 'a+')
else:
print('No new repos found!')
async def run_async(repo_name: str, path: str, cmds: List[str]) -> Union[None, str]:
"""
Run `cmds` asynchronously in `path` directory. Return the `path` if
execution fails.
"""
process = await asyncio.create_subprocess_exec(
*cmds,
stdin=asyncio.subprocess.DEVNULL,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
start_new_session=True,
cwd=path)
stdout, stderr = await process.communicate()
for pipe in (stdout, stderr):
if pipe:
print(format_output(pipe.decode(), f'{repo_name}: '))
# The existence of stderr is not good indicator since git sometimes write
# to stderr even if the execution is successful, e.g. git fetch
if process.returncode != 0:
return path
def format_output(s: str, prefix: str):
"""
Prepends every line in given string with the given prefix.
"""
return ''.join([f'{prefix}{line}' for line in s.splitlines(keepends=True)])
def exec_async_tasks(tasks: List[Coroutine]) -> List[Union[None, str]]:
"""
Execute tasks asynchronously
"""
# TODO: asyncio API is nicer in python 3.7
if platform.system() == 'Windows':
loop = asyncio.ProactorEventLoop()
asyncio.set_event_loop(loop)
else:
loop = asyncio.get_event_loop()
try:
errors = loop.run_until_complete(asyncio.gather(*tasks))
finally:
loop.close()
return errors
def describe(repos: Dict[str, str]) -> str:
"""
Return the status of all repos
"""
if repos:
name_width = max(len(n) for n in repos) + 1
funcs = info.get_info_funcs()
for name in sorted(repos):
path = repos[name]
display_items = ' '.join(f(path) for f in funcs)
yield f'{name:<{name_width}}{display_items}'
def get_cmds_from_files() -> Dict[str, Dict[str, str]]:
"""
Parse delegated git commands from default config file
and custom config file.
Example return
{
'branch': {'help': 'show local branches'},
'clean': {'cmd': 'clean -dfx',
'help': 'remove all untracked files/folders'},
}
"""
# default config file
fname = os.path.join(os.path.dirname(__file__), "cmds.yml")
with open(fname, 'r') as stream:
cmds = yaml.load(stream, Loader=yaml.FullLoader)
# custom config file
root = common.get_config_dir()
fname = os.path.join(root, 'cmds.yml')
custom_cmds = {}
if os.path.isfile(fname) and os.path.getsize(fname):
with open(fname, 'r') as stream:
custom_cmds = yaml.load(stream, Loader=yaml.FullLoader)
# custom commands shadow default ones
cmds.update(custom_cmds)
return cmds