1
0
Fork 0

Adding upstream version 0.11.9.

Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
Daniel Baumann 2025-02-11 18:39:36 +01:00
parent 1b2b356ce0
commit 3ac0de4543
Signed by: daniel
GPG key ID: FBB4F0E80A80222F
11 changed files with 528 additions and 157 deletions

View file

@ -2,19 +2,23 @@ import os
import yaml
import asyncio
import platform
from functools import lru_cache
from functools import lru_cache, partial
from pathlib import Path
from typing import List, Dict, Coroutine, Union
from . import info
from . import common
def get_config_fname(fname: str) -> str:
@lru_cache()
def get_context() -> Union[Path, None]:
"""
Return the file name that stores the repo locations.
Return the context: either a group name or 'none'
"""
root = common.get_config_dir()
return os.path.join(root, fname)
config_dir = Path(common.get_config_dir())
matches = list(config_dir.glob('*.context'))
assert len(matches) < 2, "Cannot have multiple .context file"
return matches[0] if matches else None
@lru_cache()
@ -22,7 +26,7 @@ def get_repos() -> Dict[str, str]:
"""
Return a `dict` of repo name to repo absolute path
"""
path_file = get_config_fname('repo_path')
path_file = common.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:
@ -47,7 +51,7 @@ def get_groups() -> Dict[str, List[str]]:
"""
Return a `dict` of group name to repo names.
"""
fname = get_config_fname('groups.yml')
fname = common.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:
@ -102,7 +106,7 @@ 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')
fname = common.get_config_fname('repo_path')
os.makedirs(os.path.dirname(fname), exist_ok=True)
with open(fname, mode) as f:
f.write(data)
@ -112,10 +116,13 @@ def write_to_groups_file(groups: Dict[str, List[str]], mode: str):
"""
"""
fname = get_config_fname('groups.yml')
fname = common.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)
if not groups: # all groups are deleted
open(fname, 'w').close()
else:
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]):
@ -182,17 +189,23 @@ def exec_async_tasks(tasks: List[Coroutine]) -> List[Union[None, str]]:
return errors
def describe(repos: Dict[str, str]) -> str:
def describe(repos: Dict[str, str], no_colors: bool=False) -> str:
"""
Return the status of all repos
"""
if repos:
name_width = max(len(n) for n in repos) + 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)
for name in sorted(repos):
path = repos[name]
display_items = ' '.join(f(path) for f in funcs)
yield f'{name:<{name_width}}{display_items}'
info_items = ' '.join(f(path) for f in funcs)
yield f'{name:<{name_width}}{info_items}'
def get_cmds_from_files() -> Dict[str, Dict[str, str]]: