mirror of
https://github.com/openSUSE/osc.git
synced 2026-03-09 10:35:52 +01:00
Merge pull request #2035 from dmach/osc-search-scmsync
Extend 'osc search' with gitea data cached in an external service
This commit is contained in:
GitHub
@@ -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