From 1639cd84e1fa0c7e432d2faed36edcd519f83b8c Mon Sep 17 00:00:00 2001 From: Daniel Mach Date: Thu, 29 Jan 2026 08:55:46 +0100 Subject: [PATCH] Extend 'osc search' with gitea data cached in an external service The service name is currently derived from apiurl, the hostname is 'packages'. If the service is not reachable, the error is ignored. The output of 'osc search' is now prefixed with [obs] or [git] to clarify origin of the data. --- osc/commandline.py | 116 +++++++++++++++++++++++++++++++++++ osc/gitea_api/cache.py | 134 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 250 insertions(+) create mode 100644 osc/gitea_api/cache.py 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()