diff mbox series

[1/1] support/scripts/pkg-stats: configurable upstream-check api endpoint

Message ID 20250925045115.1655-1-git@jdknight.me
State New
Headers show
Series [1/1] support/scripts/pkg-stats: configurable upstream-check api endpoint | expand

Commit Message

James Knight Sept. 25, 2025, 4:51 a.m. UTC 
Provide the ability for a caller to override the API endpoint used for
upstream checks. This can be helpful for environments which support a
reverse proxy for caching an Anitya target.

Signed-off-by: James Knight <git@jdknight.me>
---
 support/scripts/pkg-stats | 31 +++++++++++++++++++------------
 1 file changed, 19 insertions(+), 12 deletions(-)
diff mbox series

Patch

diff --git a/support/scripts/pkg-stats b/support/scripts/pkg-stats
index a0a46aa8cbac1532f84d67bed9924dace4f6778b..3b9c9bd5b11d6434ffab847fe5de18154936e34e 100755
--- a/support/scripts/pkg-stats
+++ b/support/scripts/pkg-stats
@@ -37,6 +37,7 @@  from getdeveloperlib import parse_developers  # noqa: E402
 INFRA_RE = re.compile(r"\$\(eval \$\(([a-z-]*)-package\)\)")
 URL_RE = re.compile(r"\s*https?://\S*\s*$")
 
+DEFAULT_RM_API_ENDPOINT = "https://release-monitoring.org/api"
 RM_API_STATUS_ERROR = 1
 RM_API_STATUS_FOUND_BY_DISTRO = 2
 RM_API_STATUS_FOUND_BY_PATTERN = 3
@@ -523,8 +524,8 @@  def check_package_latest_version_set_status(pkg, status, version, identifier):
         pkg.status['version'] = ('ok', 'up-to-date')
 
 
-async def check_package_get_latest_version_by_distro(session, pkg, retry=True):
-    url = "https://release-monitoring.org/api/project/Buildroot/%s" % pkg.name
+async def check_package_get_latest_version_by_distro(rm_url, session, pkg, retry=True):
+    url = f"{rm_url}/project/Buildroot/{pkg.name}"
     try:
         async with session.get(url) as resp:
             if resp.status != 200:
@@ -545,13 +546,13 @@  async def check_package_get_latest_version_by_distro(session, pkg, retry=True):
 
     except (aiohttp.ClientError, asyncio.TimeoutError):
         if retry:
-            return await check_package_get_latest_version_by_distro(session, pkg, retry=False)
+            return await check_package_get_latest_version_by_distro(rm_url, session, pkg, retry=False)
         else:
             return False
 
 
-async def check_package_get_latest_version_by_guess(session, pkg, retry=True):
-    url = "https://release-monitoring.org/api/projects/?pattern=%s" % pkg.name
+async def check_package_get_latest_version_by_guess(rm_url, session, pkg, retry=True):
+    url = f"{rm_url}/projects/?pattern={pkg.name}"
     try:
         async with session.get(url) as resp:
             if resp.status != 200:
@@ -576,7 +577,7 @@  async def check_package_get_latest_version_by_guess(session, pkg, retry=True):
 
     except (aiohttp.ClientError, asyncio.TimeoutError):
         if retry:
-            return await check_package_get_latest_version_by_guess(session, pkg, retry=False)
+            return await check_package_get_latest_version_by_guess(rm_url, session, pkg, retry=False)
         else:
             return False
 
@@ -584,16 +585,16 @@  async def check_package_get_latest_version_by_guess(session, pkg, retry=True):
 check_latest_count = 0
 
 
-async def check_package_latest_version_get(session, pkg, npkgs, verbose=False):
+async def check_package_latest_version_get(rm_url, session, pkg, npkgs, verbose=False):
     global check_latest_count
 
-    if await check_package_get_latest_version_by_distro(session, pkg):
+    if await check_package_get_latest_version_by_distro(rm_url, session, pkg):
         check_latest_count += 1
         if verbose:
             print("[%04d/%04d] %s" % (check_latest_count, npkgs, pkg.name))
         return
 
-    if await check_package_get_latest_version_by_guess(session, pkg):
+    if await check_package_get_latest_version_by_guess(rm_url, session, pkg):
         check_latest_count += 1
         if verbose:
             print("[%04d/%04d] %s" % (check_latest_count, npkgs, pkg.name))
@@ -607,7 +608,7 @@  async def check_package_latest_version_get(session, pkg, npkgs, verbose=False):
         print("[%04d/%04d] %s" % (check_latest_count, npkgs, pkg.name))
 
 
-async def check_package_latest_version(packages, verbose=False):
+async def check_package_latest_version(rm_url, packages, verbose=False):
     """
     Fills in the .latest_version field of all Package objects
 
@@ -630,7 +631,7 @@  async def check_package_latest_version(packages, verbose=False):
     async with aiohttp.ClientSession(connector=connector, trust_env=True) as sess:
         packages = [p for p in packages if p.is_actual_package]
         for pkg in packages:
-            tasks.append(asyncio.ensure_future(check_package_latest_version_get(sess, pkg, len(packages), verbose=verbose)))
+            tasks.append(asyncio.ensure_future(check_package_latest_version_get(rm_url, sess, pkg, len(packages), verbose=verbose)))
         await asyncio.wait(tasks)
 
 
@@ -1303,6 +1304,9 @@  def parse_args():
                           help='List of packages (comma separated)')
     parser.add_argument('--nvd-path', dest='nvd_path',
                         help='Path to the local NVD database', type=resolvepath)
+    parser.add_argument('--upstream-url', dest='upstream_url',
+                        help='Upstream API endpoint to use',
+                        default=DEFAULT_RM_API_ENDPOINT)
     parser.add_argument('--disable', type=list_str,
                         help='Features to disable, comma-separated (cve, upstream, url, warnings)',
                         default=[])
@@ -1317,6 +1321,9 @@  def parse_args():
 def __main__():
     global cvecheck
 
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+
     args = parse_args()
 
     if args.nvd_path:
@@ -1365,7 +1372,7 @@  def __main__():
     if "upstream" not in args.disable:
         print("Getting latest versions ...")
         loop = asyncio.get_event_loop()
-        loop.run_until_complete(check_package_latest_version(packages, verbose=args.verbose))
+        loop.run_until_complete(check_package_latest_version(args.upstream_url, packages, verbose=args.verbose))
     if "cve" not in args.disable and args.nvd_path:
         print("Checking packages CVEs")
         check_package_cves(args.nvd_path, packages)