mirror of
https://github.com/openSUSE/osc.git
synced 2026-03-08 01:56:15 +01:00
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.
This commit is contained in:
@@ -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',
|
||||
|
||||
134
osc/gitea_api/cache.py
Normal file
134
osc/gitea_api/cache.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user