diff --git a/osc/commandline.py b/osc/commandline.py index 911f3e37..a1dcdcd3 100644 --- a/osc/commandline.py +++ b/osc/commandline.py @@ -9229,6 +9229,10 @@ Please submit there instead, or use --nodevelproject to force direct submission. from .core import get_source_rev from .core import search from .core import xpath_join + from .gitea_api.cache import gitea_cache_search_projects + from .gitea_api.cache import gitea_cache_search_packages + from .gitea_api.cache import gitea_cache_search_project_maintainers + from .gitea_api.cache import gitea_cache_search_package_maintainers def build_xpath(attr, what, substr=False): if substr: @@ -9397,6 +9401,7 @@ Please submit there instead, or use --nodevelproject to force direct submission. result.append(node.get('filepath')) results.append(result) + if not results: print(f'No matches found for \'{role_filter or search_term}\' in {kind}s') continue @@ -9436,6 +9441,117 @@ Please submit there instead, or use --nodevelproject to force direct submission. for row in build_table(len(headline), results, headline, 2, csv=opts.csv): print(row) + if not any((opts.project, opts.package)): + opts.project = True + opts.package = True + + def print_projects(entries): + headline = ["# Project"] + if opts.verbose: + headline += ["# Git URL", "# Branch"] + if opts.version: + headline += ["# Commit"] + results = [] + for i in entries: + # unwrap project when searching via maintainer + if "project" in i: + i = i["project"] + result = [] + result.append(i["name"]) # project + if opts.verbose: + result.append(i["git_url"]) + result.append(i["git_branch"]) + if opts.version: + result.append(i["git_commit"]) + results.append(result) + + results.sort(key=itemgetter(0)) + results = list(itertools.chain.from_iterable(results)) + for row in build_table(len(headline), results, headline, 2, csv=opts.csv): + print(row) + + def print_packages(entries): + headline = ["# Project", "# Package"] + if opts.verbose: + headline += ["# Git URL", "# Branch"] + if opts.version: + headline += ["# Commit"] + results = [] + for i in entries: + # unwrap package when searching via maintainer + if "package" in i: + i = i["package"] + result = [] + result.append(i["project"]["name"]) # project + result.append(i["name"]) # package + if opts.verbose: + result.append(i["git_url"]) + result.append(i["git_branch"]) + if opts.version: + result.append(i["git_commit"]) + results.append(result) + + results.sort(key=itemgetter(0, 1)) + results = list(itertools.chain.from_iterable(results)) + for row in build_table(len(headline), results, headline, 2, csv=opts.csv): + print(row) + + # query a new service that caches various gitea information + if opts.maintainer or opts.bugowner or opts.involved: + if opts.project: + q = {} + if opts.substring: + q["users__like"] = [search_term] + else: + q["users"] = [search_term] + + results = gitea_cache_search_project_maintainers(**q) + if results: + if not opts.csv: + print(f"\n[git] matches for '{search_term}' in project maintainers:\n") + print_projects(results) + + if opts.package: + q = {} + if opts.substring: + q["users__like"] = [search_term] + else: + q["users"] = [search_term] + + results = gitea_cache_search_package_maintainers(**q) + if results: + if not opts.csv: + print(f"\n[git] matches for '{search_term}' in package maintainers:\n") + print_packages(results) + + else: + if opts.project: + q = {} + if opts.substring: + q["names__like"] = [search_term] + else: + q["names"] = [search_term] + + results = gitea_cache_search_projects(**q) + if results: + if not opts.csv: + print(f"\n[git] matches for '{search_term}' in projects:\n") + print_projects(results) + + if opts.package: + q = {} + if opts.substring: + q["names__like"] = [search_term] + else: + q["names"] = [search_term] + + results = gitea_cache_search_packages(**q) + if results: + if not opts.csv: + print(f"\n[git] matches for '{search_term}' in packages:\n") + print_packages(results) + + @cmdln.option('-p', '--project', metavar='project', help='specify the path to a project') @cmdln.option('-n', '--name', metavar='name', diff --git a/osc/gitea_api/cache.py b/osc/gitea_api/cache.py new file mode 100644 index 00000000..9fec3657 --- /dev/null +++ b/osc/gitea_api/cache.py @@ -0,0 +1,134 @@ +import functools +from typing import List +from typing import Optional + + +def get_default_base_url(): + from ..conf import config + + result = config.apiurl.replace("://api.", "://packages.") + return result + + +def ignore_http_errors(func): + """ + Return [] if `base_url` host is not found or doesn't return the expected status. + This is needed because majority of OBS deployments don't have the new service for searching. + """ + + @functools.wraps(func) + def wrapper(*args, **kwargs): + from urllib3.exceptions import NameResolutionError, MaxRetryError + + try: + response = func(*args, **kwargs) + + if hasattr(response, "status") and 400 <= response.status < 500: + return [] + + return response + + except (NameResolutionError, MaxRetryError): + return [] + + return wrapper + + +@ignore_http_errors +def gitea_cache_search_packages( + base_url: Optional[str] = None, + names: Optional[List[str]] = None, + names__like: Optional[List[str]] = None, + projects: Optional[List[str]] = None, + projects__like: Optional[List[str]] = None, +): + from ..core import http_request + from ..core import makeurl + + if not base_url: + base_url = get_default_base_url() + + q = { + "name": names, + "name__like": names__like, + "project__name": projects, + "project__name__like": projects__like, + } + url = makeurl(base_url, ["api", "v1", "package", "search"], q) + response = http_request("GET", url) + return response.json() + + +@ignore_http_errors +def gitea_cache_search_projects( + base_url: Optional[str] = None, + names: Optional[List[str]] = None, + names__like: Optional[List[str]] = None, + packages: Optional[List[str]] = None, + packages__like: Optional[List[str]] = None, +): + from ..core import http_request + from ..core import makeurl + + if not base_url: + base_url = get_default_base_url() + + q = { + "name": names, + "name__like": names__like, + "packages__name": packages, + "packages__name__like": packages__like, + } + url = makeurl(base_url, ["api", "v1", "project", "search"], q) + response = http_request("GET", url) + return response.json() + + +@ignore_http_errors +def gitea_cache_search_package_maintainers( + base_url: Optional[str] = None, + users: Optional[List[str]] = None, + users__like: Optional[List[str]] = None, + packages: Optional[List[str]] = None, + packages__like: Optional[List[str]] = None, +): + from ..core import http_request + from ..core import makeurl + + if not base_url: + base_url = get_default_base_url() + + q = { + "user": users, + "user__like": users__like, + "packages__name": packages, + "packages__name__like": packages__like, + } + url = makeurl(base_url, ["api", "v1", "package", "maintainer", "search"], q) + response = http_request("GET", url) + return response.json() + + +@ignore_http_errors +def gitea_cache_search_project_maintainers( + base_url: Optional[str] = None, + users: Optional[List[str]] = None, + users__like: Optional[List[str]] = None, + projects: Optional[List[str]] = None, + projects__like: Optional[List[str]] = None, +): + from ..core import http_request + from ..core import makeurl + + if not base_url: + base_url = get_default_base_url() + + q = { + "user": users, + "user__like": users__like, + "project__name": projects, + "project__name__like": projects__like, + } + url = makeurl(base_url, ["api", "v1", "project", "maintainer", "search"], q) + response = http_request("GET", url) + return response.json()