git/gita
1
0
Fork 0
Code Issues Activity

Merging upstream version 0.12.9.

Browse source 
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
Daniel Baumann 2025-02-11 18:41:20 +01:00
parent f0c873f1f0
commit cdd1ab6dcf
Signed by: daniel
GPG key ID: FBB4F0E80A80222F
8 changed files with 252 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

135
gita/__main__.py
View file

@ -44,7 +44,11 @@ def f_color(args: argparse.Namespace):
if cmd == 'll': # pragma: no cover
info.show_colors()
elif cmd == 'set':
print('not implemented')
colors = info.get_color_encoding()
colors[args.situation] = info.Color[args.color].value
yml_config = common.get_config_fname('color.yml')
with open(yml_config, 'w') as f:
yaml.dump(colors, f, default_flow_style=None)
def f_info(args: argparse.Namespace):
@ -68,6 +72,23 @@ def f_info(args: argparse.Namespace):
yaml.dump(to_display, f, default_flow_style=None)
def f_clone(args: argparse.Namespace):
path = Path.cwd()
errors = utils.exec_async_tasks(
utils.run_async(repo_name, path, ['git', 'clone', url])
for url, repo_name, _ in utils.parse_clone_config(args.fname))
def f_freeze(_):
repos = utils.get_repos()
for name, path in repos.items():
url = ''
cp = subprocess.run(['git', 'remote', '-v'], cwd=path, capture_output=True)
if cp.returncode == 0:
url = cp.stdout.decode('utf-8').split('\n')[0].split()[1]
print(f'{url},{name},{path}')
def f_ll(args: argparse.Namespace):
"""
Display details of all repos
@ -108,8 +129,11 @@ def f_group(args: argparse.Namespace):
del groups[gname]
utils.write_to_groups_file(groups, 'w')
elif cmd == 'rm':
ctx = utils.get_context()
for name in args.to_ungroup:
del groups[name]
if ctx and str(ctx.stem) == name:
ctx.unlink()
utils.write_to_groups_file(groups, 'w')
elif cmd == 'add':
gname = args.gname
@ -120,6 +144,15 @@ def f_group(args: argparse.Namespace):
utils.write_to_groups_file(groups, 'w')
else:
utils.write_to_groups_file({gname: sorted(args.to_group)}, 'a+')
elif cmd == 'rmrepo':
gname = args.gname
if gname in groups:
for repo in args.from_group:
try:
groups[gname].remove(repo)
except ValueError as e:
pass
utils.write_to_groups_file(groups, 'w')
def f_context(args: argparse.Namespace):
@ -189,6 +222,42 @@ def f_git_cmd(args: argparse.Namespace):
subprocess.run(cmds, cwd=path)
def f_shell(args):
"""
Delegate shell command defined in `args.man`, which may or may not
contain repo names.
"""
names = []
repos = utils.get_repos()
groups = utils.get_groups()
ctx = utils.get_context()
for i, word in enumerate(args.man):
if word in repos or word in groups:
names.append(word)
else:
break
args.repo = names
# TODO: redundant with f_git_cmd
if not args.repo and ctx:
args.repo = [ctx.stem]
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 = args.man[i:]
for name, path in repos.items():
# TODO: pull this out as a function
got = subprocess.run(cmds, cwd=path, check=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
print(utils.format_output(got.stdout.decode(), name))
def f_super(args):
"""
Delegate git command/alias defined in `args.man`, which may or may not
@ -221,32 +290,40 @@ def main(argv=None):
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 = subparsers.add_parser('add', description='add repo(s)',
help='add repo(s)')
p_add.add_argument('paths', nargs='+', help="repo(s) to add")
p_add.add_argument('-r', dest='recursive', action='store_true',
help="recursively add repo(s) in the given path.")
p_add.set_defaults(func=f_add)
p_rm = subparsers.add_parser('rm', help='remove repo(s)')
p_rm = subparsers.add_parser('rm', description='remove repo(s)',
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_freeze = subparsers.add_parser('freeze', description='print all repo information')
p_freeze.set_defaults(func=f_freeze)
p_clone = subparsers.add_parser('clone', description='clone repos from config file')
p_clone.add_argument('fname',
help='config file. Its content should be the output of `gita freeze`.')
p_clone.set_defaults(func=f_clone)
p_rename = subparsers.add_parser('rename', description='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.add_argument('new_name', help="new name")
p_rename.set_defaults(func=f_rename)
p_color = subparsers.add_parser('color',
help='display and modify branch coloring of the ll sub-command.')
description='display and modify branch coloring of the ll sub-command.')
p_color.set_defaults(func=f_color)
color_cmds = p_color.add_subparsers(dest='color_cmd',
help='additional help with sub-command -h')
@ -262,7 +339,7 @@ def main(argv=None):
help="available colors")
p_info = subparsers.add_parser('info',
help='list, add, or remove information items of the ll sub-command.')
description='list, add, or remove information items of the ll sub-command.')
p_info.set_defaults(func=f_info)
info_cmds = p_info.add_subparsers(dest='info_cmd',
help='additional help with sub-command -h')
@ -297,12 +374,12 @@ def main(argv=None):
nargs='?',
choices=utils.get_groups(),
help="show repos in the chosen group")
p_ll.add_argument('-n', '--no-colors', action='store_true',
p_ll.add_argument('-C', '--no-colors', action='store_true',
help='Disable coloring on the branch names.')
p_ll.set_defaults(func=f_ll)
p_context = subparsers.add_parser('context',
help='Set and remove context. A context is a group.'
description='Set and remove context. A context is a group.'
' When set, all operations apply only to repos in that group.')
p_context.add_argument('choice',
nargs='?',
@ -311,7 +388,7 @@ def main(argv=None):
p_context.set_defaults(func=f_context)
p_ls = subparsers.add_parser(
'ls', help='display names of all repos, or path of a chosen repo')
'ls', description='display names of all repos, or path of a chosen repo')
p_ls.add_argument('repo',
nargs='?',
choices=utils.get_repos(),
@ -319,7 +396,7 @@ def main(argv=None):
p_ls.set_defaults(func=f_ls)
p_group = subparsers.add_parser(
'group', help='list, add, or remove repo group(s)')
'group', description='list, add, or remove repo group(s)')
p_group.set_defaults(func=f_group)
group_cmds = p_group.add_subparsers(dest='group_cmd',
help='additional help with sub-command -h')
@ -336,6 +413,17 @@ def main(argv=None):
metavar='group-name',
required=True,
help="group name")
pg_rmrepo = group_cmds.add_parser('rmrepo', description='remove repo(s) from a group.')
pg_rmrepo.add_argument('from_group',
nargs='+',
metavar='repo',
choices=utils.get_repos(),
help="repo(s) to be removed from the group")
pg_rmrepo.add_argument('-n', '--name',
dest='gname',
metavar='group-name',
required=True,
help="group name")
pg_rename = group_cmds.add_parser('rename', description='Change group name.')
pg_rename.add_argument('gname', metavar='group-name',
choices=utils.get_groups(),
@ -351,7 +439,7 @@ def main(argv=None):
# superman mode
p_super = subparsers.add_parser(
'super',
help='superman mode: delegate any git command/alias in specified or '
description='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')
@ -363,6 +451,21 @@ def main(argv=None):
"Another: gita super checkout master ")
p_super.set_defaults(func=f_super)
# shell mode
p_shell = subparsers.add_parser(
'shell',
description='shell mode: delegate any shell command in specified or '
'all repo(s).\n'
'Examples:\n \t gita shell pwd\n'
'\t gita shell repo1 repo2 repo3 touch xx')
p_shell.add_argument(
'man',
nargs=argparse.REMAINDER,
help="execute arbitrary shell command for specified or all repos "
"Example: gita shell myrepo1 ls"
"Another: gita shell git checkout master ")
p_shell.set_defaults(func=f_shell)
# sub-commands that fit boilerplate
cmds = utils.get_cmds_from_files()
for name, data in cmds.items():
@ -376,7 +479,7 @@ def main(argv=None):
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 = subparsers.add_parser(name, description=help)
sp.add_argument('repo', nargs=nargs, choices=choices, help=help)
sp.set_defaults(func=f_git_cmd, cmd=cmd.split())

37
gita/info.py
View file

@ -37,29 +37,36 @@ def show_colors(): # pragma: no cover
"""
"""
names = {c.value: c.name for c in Color}
for i, c in enumerate(Color, start=1):
if c != Color.end:
print(f'{c.value}{c.name:<8} ', end='')
if i % 9 == 0:
print()
print(f'{Color.end}')
for situation, c in get_color_encoding().items():
print(f'{situation:<12}: {c.value}{c.name:<8}{Color.end} ')
for situation, c in sorted(get_color_encoding().items()):
print(f'{situation:<12}: {c}{names[c]:<8}{Color.end} ')
@lru_cache()
def get_color_encoding():
def get_color_encoding() -> Dict[str, str]:
"""
Return color scheme for different local/remote situations.
"""
# TODO: add config file
return {
'no-remote': Color.white,
'in-sync': Color.green,
'diverged': Color.red,
'local-ahead': Color.purple,
'remote-ahead': Color.yellow,
# custom settings
yml_config = Path(common.get_config_fname('color.yml'))
if yml_config.is_file():
with open(yml_config, 'r') as stream:
colors = yaml.load(stream, Loader=yaml.FullLoader)
else:
colors = {
'no-remote': Color.white.value,
'in-sync': Color.green.value,
'diverged': Color.red.value,
'local-ahead': Color.purple.value,
'remote-ahead': Color.yellow.value,
}
return colors
def get_info_funcs() -> List[Callable[[str], str]]:
@ -82,20 +89,20 @@ def get_info_items() -> List[str]:
"""
Return the information items to be displayed in the `gita ll` command.
"""
# default settings
display_items = ['branch', 'commit_msg']
# custom settings
yml_config = Path(common.get_config_fname('info.yml'))
if yml_config.is_file():
with open(yml_config, 'r') as stream:
display_items = yaml.load(stream, Loader=yaml.FullLoader)
display_items = [x for x in display_items if x in ALL_INFO_ITEMS]
else:
# default settings
display_items = ['branch', 'commit_msg']
return display_items
def get_path(path):
return Color.cyan + path + Color.end
return f'{Color.cyan}{path}{Color.end}'
def get_head(path: str) -> str:

18
gita/utils.py
View file

@ -4,7 +4,7 @@ import asyncio
import platform
from functools import lru_cache, partial
from pathlib import Path
from typing import List, Dict, Coroutine, Union
from typing import List, Dict, Coroutine, Union, Iterator
from . import info
from . import common
@ -60,7 +60,6 @@ def get_groups() -> Dict[str, List[str]]:
return groups
def get_choices() -> List[Union[str, None]]:
"""
Return all repo names, group names, and an additional empty list. The empty
@ -128,6 +127,8 @@ def write_to_groups_file(groups: Dict[str, List[str]], mode: str):
def add_repos(repos: Dict[str, str], new_paths: List[str]):
"""
Write new repo paths to file
@param repos: name -> path
"""
existing_paths = set(repos.values())
new_paths = set(os.path.abspath(p) for p in new_paths if is_git(p))
@ -142,6 +143,15 @@ def add_repos(repos: Dict[str, str], new_paths: List[str]):
print('No new repos found!')
def parse_clone_config(fname: str) -> Iterator[List[str]]:
"""
Return the url, name, and path of all repos in `fname`.
"""
with open(fname) as f:
for line in f:
yield line.strip().split(',')
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
@ -157,7 +167,7 @@ async def run_async(repo_name: str, path: str, cmds: List[str]) -> Union[None, s
stdout, stderr = await process.communicate()
for pipe in (stdout, stderr):
if pipe:
print(format_output(pipe.decode(), f'{repo_name}: '))
print(format_output(pipe.decode(), 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:
@ -168,7 +178,7 @@ 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)])
return ''.join([f'{prefix}: {line}' for line in s.splitlines(keepends=True)])
def exec_async_tasks(tasks: List[Coroutine]) -> List[Union[None, str]]: