git/gita
1
0
Fork 0
Code Issues Activity

Merging upstream version 0.16.4.

Browse source 
Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
Daniel Baumann 2025-02-11 18:50:00 +01:00
parent 9c2ea8d556
commit c872e9becf
Signed by: daniel
GPG key ID: FBB4F0E80A80222F
11 changed files with 143 additions and 124 deletions
Download patch file Download diff file Expand all files Collapse all files

2
gita/__init__.py
View file

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

7
gita/common.py
View file

@ -2,8 +2,11 @@ import os
def get_config_dir() -> str:
root = os.environ.get('GITA_PROJECT_HOME') or os.environ.get('XDG_CONFIG_HOME') or os.path.join(
os.path.expanduser('~'), '.config')
root = (
os.environ.get("GITA_PROJECT_HOME")
or os.environ.get("XDG_CONFIG_HOME")
or os.path.join(os.path.expanduser("~"), ".config")
)
return os.path.join(root, "gita")

202
gita/info.py
View file

@ -3,7 +3,7 @@ import csv
import subprocess
from enum import Enum
from pathlib import Path
from functools import lru_cache
from functools import lru_cache, partial
from typing import Tuple, List, Callable, Dict
from . import common
@ -13,24 +13,25 @@ class Color(Enum):
"""
Terminal color
"""
black = '\x1b[30m'
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'
b_black = '\x1b[30;1m'
b_red = '\x1b[31;1m'
b_green = '\x1b[32;1m'
b_yellow = '\x1b[33;1m'
b_blue = '\x1b[34;1m'
b_purple = '\x1b[35;1m'
b_cyan = '\x1b[36;1m'
b_white = '\x1b[37;1m'
underline = '\x1B[4m'
black = "\x1b[30m"
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"
b_black = "\x1b[30;1m"
b_red = "\x1b[31;1m"
b_green = "\x1b[32;1m"
b_yellow = "\x1b[33;1m"
b_blue = "\x1b[34;1m"
b_purple = "\x1b[35;1m"
b_cyan = "\x1b[36;1m"
b_white = "\x1b[37;1m"
underline = "\x1B[4m"
# Make f"{Color.foo}" expand to Color.foo.value .
#
@ -40,26 +41,24 @@ class Color(Enum):
default_colors = {
'no-remote': Color.white.name,
'in-sync': Color.green.name,
'diverged': Color.red.name,
'local-ahead': Color.purple.name,
'remote-ahead': Color.yellow.name,
}
"no-remote": Color.white.name,
"in-sync": Color.green.name,
"diverged": Color.red.name,
"local-ahead": Color.purple.name,
"remote-ahead": Color.yellow.name,
}
def show_colors(): # pragma: no cover
"""
"""
""" """
for i, c in enumerate(Color, start=1):
if c != Color.end and c != Color.underline:
print(f'{c.value}{c.name:<8} ', end='')
print(f"{c.value}{c.name:<8} ", end="")
if i % 9 == 0:
print()
print(f'{Color.end}')
print(f"{Color.end}")
for situation, c in sorted(get_color_encoding().items()):
print(f'{situation:<12}: {Color[c].value}{c:<8}{Color.end} ')
print(f"{situation:<12}: {Color[c].value}{c:<8}{Color.end} ")
@lru_cache()
@ -69,9 +68,9 @@ def get_color_encoding() -> Dict[str, str]:
In the format of {situation: color name}
"""
# custom settings
csv_config = Path(common.get_config_fname('color.csv'))
csv_config = Path(common.get_config_fname("color.csv"))
if csv_config.is_file():
with open(csv_config, 'r') as f:
with open(csv_config, "r") as f:
reader = csv.DictReader(f)
colors = next(reader)
else:
@ -79,7 +78,7 @@ def get_color_encoding() -> Dict[str, str]:
return colors
def get_info_funcs() -> List[Callable[[str], str]]:
def get_info_funcs(no_colors=False) -> 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.
@ -88,11 +87,12 @@ def get_info_funcs() -> List[Callable[[str], str]]:
to_display = get_info_items()
# This re-definition is to make unit test mocking to work
all_info_items = {
'branch': get_repo_status,
'commit_msg': get_commit_msg,
'commit_time': get_commit_time,
'path': get_path,
}
"branch": partial(get_repo_status, no_colors=no_colors),
"branch_name": get_repo_branch,
"commit_msg": get_commit_msg,
"commit_time": get_commit_time,
"path": get_path,
}
return [all_info_items[k] for k in to_display]
@ -101,15 +101,15 @@ def get_info_items() -> List[str]:
Return the information items to be displayed in the `gita ll` command.
"""
# custom settings
csv_config = Path(common.get_config_fname('info.csv'))
csv_config = Path(common.get_config_fname("info.csv"))
if csv_config.is_file():
with open(csv_config, 'r') as f:
with open(csv_config, "r") as f:
reader = csv.reader(f)
display_items = next(reader)
display_items = [x for x in display_items if x in ALL_INFO_ITEMS]
else:
# default settings
display_items = ['branch', 'commit_msg', 'commit_time']
display_items = ["branch", "commit_msg", "commit_time"]
return display_items
@ -119,43 +119,48 @@ def get_path(prop: Dict[str, str]) -> str:
# TODO: do we need to add the flags here too?
def get_head(path: str) -> str:
result = subprocess.run('git symbolic-ref -q --short HEAD || git describe --tags --exact-match',
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
universal_newlines=True,
cwd=path)
result = subprocess.run(
"git symbolic-ref -q --short HEAD || git describe --tags --exact-match",
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
universal_newlines=True,
cwd=path,
)
return result.stdout.strip()
def run_quiet_diff(flags: List[str], args: List[str]) -> int:
def run_quiet_diff(flags: List[str], args: List[str], path) -> int:
"""
Return the return code of git diff `args` in quiet mode
"""
result = subprocess.run(
['git'] + flags + ['diff', '--quiet'] + args,
["git"] + flags + ["diff", "--quiet"] + args,
stderr=subprocess.DEVNULL,
cwd=path,
)
return result.returncode
def get_common_commit() -> str:
def get_common_commit(path) -> 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)
result = subprocess.run(
"git merge-base @{0} @{u}".split(),
stdout=subprocess.PIPE,
universal_newlines=True,
cwd=path,
)
return result.stdout.strip()
def has_untracked(flags: List[str]) -> bool:
def has_untracked(flags: List[str], path) -> bool:
"""
Return True if untracked file/folder exists
"""
cmd = ['git'] + flags + 'ls-files -zo --exclude-standard'.split()
result = subprocess.run(cmd,
stdout=subprocess.PIPE)
cmd = ["git"] + flags + "ls-files -zo --exclude-standard".split()
result = subprocess.run(cmd, stdout=subprocess.PIPE, cwd=path)
return bool(result.stdout)
@ -164,12 +169,14 @@ def get_commit_msg(prop: Dict[str, str]) -> str:
Return the last commit message.
"""
# `git show-branch --no-name HEAD` is faster than `git show -s --format=%s`
cmd = ['git'] + prop['flags'] + 'show-branch --no-name HEAD'.split()
result = subprocess.run(cmd,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
universal_newlines=True,
cwd=prop['path'])
cmd = ["git"] + prop["flags"] + "show-branch --no-name HEAD".split()
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
universal_newlines=True,
cwd=prop["path"],
)
return result.stdout.strip()
@ -177,60 +184,65 @@ def get_commit_time(prop: Dict[str, str]) -> str:
"""
Return the last commit time in parenthesis.
"""
cmd = ['git'] + prop['flags'] + 'log -1 --format=%cd --date=relative'.split()
result = subprocess.run(cmd,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
universal_newlines=True,
cwd=prop['path'])
cmd = ["git"] + prop["flags"] + "log -1 --format=%cd --date=relative".split()
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
universal_newlines=True,
cwd=prop["path"],
)
return f"({result.stdout.strip()})"
def get_repo_status(prop: Dict[str, str], no_colors=False) -> str:
head = get_head(prop['path'])
head = get_head(prop["path"])
dirty, staged, untracked, color = _get_repo_status(prop, no_colors)
if color:
return f'{color}{head+" "+dirty+staged+untracked:<10}{Color.end}'
return f'{head+" "+dirty+staged+untracked:<10}'
return f"{color}{head+' ['+dirty+staged+untracked+']':<13}{Color.end}"
return f"{head+' ['+dirty+staged+untracked+']':<13}"
def get_repo_branch(prop: Dict[str, str]) -> str:
return get_head(prop["path"])
def _get_repo_status(prop: Dict[str, str], no_colors: bool) -> Tuple[str]:
"""
Return the status of one repo
"""
path = prop['path']
flags = prop['flags']
os.chdir(path)
dirty = '*' if run_quiet_diff(flags, []) else ''
staged = '+' if run_quiet_diff(flags, ['--cached']) else ''
untracked = '_' if has_untracked(flags) else ''
path = prop["path"]
flags = prop["flags"]
dirty = "*" if run_quiet_diff(flags, [], path) else ""
staged = "+" if run_quiet_diff(flags, ["--cached"], path) else ""
untracked = "?" if has_untracked(flags, path) else ""
if no_colors:
return dirty, staged, untracked, ''
return dirty, staged, untracked, ""
colors = {situ: Color[name].value
for situ, name in get_color_encoding().items()}
diff_returncode = run_quiet_diff(flags, ['@{u}', '@{0}'])
colors = {situ: Color[name].value for situ, name in get_color_encoding().items()}
diff_returncode = run_quiet_diff(flags, ["@{u}", "@{0}"], path)
has_no_remote = diff_returncode == 128
has_no_diff = diff_returncode == 0
if has_no_remote:
color = colors['no-remote']
color = colors["no-remote"]
elif has_no_diff:
color = colors['in-sync']
color = colors["in-sync"]
else:
common_commit = get_common_commit()
outdated = run_quiet_diff(flags, ['@{u}', common_commit])
common_commit = get_common_commit(path)
outdated = run_quiet_diff(flags, ["@{u}", common_commit], path)
if outdated:
diverged = run_quiet_diff(flags, ['@{0}', common_commit])
color = colors['diverged'] if diverged else colors['remote-ahead']
diverged = run_quiet_diff(flags, ["@{0}", common_commit], path)
color = colors["diverged"] if diverged else colors["remote-ahead"]
else: # local is ahead of remote
color = colors['local-ahead']
color = colors["local-ahead"]
return dirty, staged, untracked, color
ALL_INFO_ITEMS = {
'branch': get_repo_status,
'commit_msg': get_commit_msg,
'commit_time': get_commit_time,
'path': get_path,
}
"branch",
"branch_name",
"commit_msg",
"commit_time",
"path",
}

14
gita/utils.py
View file

@ -5,7 +5,7 @@ import csv
import asyncio
import platform
import subprocess
from functools import lru_cache, partial
from functools import lru_cache
from pathlib import Path
from typing import List, Dict, Coroutine, Union, Iterator, Tuple
from collections import Counter, defaultdict
@ -431,18 +431,14 @@ def describe(repos: Dict[str, Dict[str, str]], no_colors: bool = False) -> str:
"""
if repos:
name_width = len(max(repos, key=len)) + 1
funcs = info.get_info_funcs()
get_repo_status = info.get_repo_status
if get_repo_status in funcs and no_colors:
idx = funcs.index(get_repo_status)
funcs[idx] = partial(get_repo_status, no_colors=True)
funcs = info.get_info_funcs(no_colors=no_colors)
num_threads = min(multiprocessing.cpu_count(), len(repos))
with ThreadPoolExecutor(max_workers=num_threads) as executor:
for line in executor.map(
lambda repo: f'{repo:<{name_width}}{" ".join(f(repos[repo]) for f in funcs)}',
sorted(repos)):
lambda name: f'{name:<{name_width}}{" ".join(f(repos[name]) for f in funcs)}',
sorted(repos),
):
yield line