add a fetch_concurrently helper function

Author: furlongm

errata/sources/distros/debian.py

@@ -28,7 +28,8 @@
 from packages.utils import find_evr, get_or_create_package
 from patchman.signals import pbar_start, pbar_update
 from util import (
-    extract, fetch_content, get_setting_of_type, get_url, run_concurrently,
+    extract, fetch_concurrently, fetch_content, get_setting_of_type, get_url,
+    run_concurrently,
 )
 from util.logging import clear_forked_pbar, error_message, warning_message
 
@@ -47,7 +48,7 @@ def update_debian_errata(concurrent_processing=True, max_workers=25):
     advisories = dsas + dlas
     fetch_dscs_from_debian_package_file_maps()
     accepted_codenames = get_accepted_debian_codenames()
-    errata = parse_debian_errata(advisories, accepted_codenames)
+    errata = parse_debian_errata(advisories, accepted_codenames, concurrent_processing, max_workers)
     create_debian_errata(errata, accepted_codenames, concurrent_processing, max_workers)
 
 
@@ -109,7 +110,7 @@ def parse_debian_package_file_map(data, repo):
             parsing_dsc = False
 
 
-def parse_debian_errata(advisories, accepted_codenames):
+def parse_debian_errata(advisories, accepted_codenames, concurrent_processing=True, max_workers=25):
     """ Parse Debian DSA/DLA files for security advisories
     """
     distro_pattern = re.compile(r'^\t\[(.+?)\] - .*')
@@ -145,7 +146,7 @@ def parse_debian_errata(advisories, accepted_codenames):
                     e['packages'][release].append(None)
     # add the last one
     errata = add_errata_by_codename(errata, e, accepted_codenames)
-    fetch_debian_dsc_package_lists(dsc_fetches)
+    fetch_debian_dsc_package_lists(dsc_fetches, concurrent_processing, max_workers)
     return errata
 
 
@@ -232,13 +233,23 @@ def process_debian_erratum(erratum, accepted_codenames):
         error_message(text=exc)
 
 
-def fetch_debian_dsc_package_lists(dsc_fetches):
+def fetch_debian_dsc_package_lists(dsc_fetches, concurrent_processing=True, max_workers=25):
     """ Fetch DSC package lists with a progress bar
     """
-    pbar_start.send(sender=None, ptext=f'Fetching {len(dsc_fetches)} Debian DSC files', plen=len(dsc_fetches))
-    for i, (package, version) in enumerate(dsc_fetches):
-        fetch_debian_dsc_package_list(package, version)
-        pbar_update.send(sender=None, index=i + 1)
+    flen = len(dsc_fetches)
+    pbar_start.send(sender=None, ptext=f'Fetching {flen} Debian DSC files', plen=flen)
+    if concurrent_processing:
+        for i, _ in enumerate(fetch_concurrently(fetch_dsc_worker, dsc_fetches, max_workers)):
+            pbar_update.send(sender=None, index=i + 1)
+    else:
+        for i, (package, version) in enumerate(dsc_fetches):
+            fetch_debian_dsc_package_list(package, version)
+            pbar_update.send(sender=None, index=i + 1)
+
+
+def fetch_dsc_worker(item, session):
+    package, version = item
+    fetch_debian_dsc_package_list(package, version, session=session)
 
 
 def get_debian_dsc_package_list(package, version):
@@ -251,14 +262,14 @@ def get_debian_dsc_package_list(package, version):
         return package_list
 
 
-def fetch_debian_dsc_package_list(package, version):
+def fetch_debian_dsc_package_list(package, version, session=None):
     """ Fetch the package list from a DSC file for a given source package/version
     """
     if not DSCs.get(package) or not DSCs[package].get(version):
         warning_message(text=f'No DSC found for {package} {version}')
         return
     source_url = DSCs[package][version]['url']
-    res = get_url(source_url)
+    res = get_url(source_url, session=session)
     data = res.content
     dsc = Dsc(data.decode())
     package_list = dsc.get('package-list')

errata/utils.py

@@ -14,16 +14,11 @@
 # You should have received a copy of the GNU General Public License
 # along with Patchman. If not, see <http://www.gnu.org/licenses/>
 
-import concurrent.futures
-
-import requests
-from requests.adapters import HTTPAdapter
-
 from errata.models import Erratum
 from packages.models import PackageUpdate
 from patchman.signals import pbar_start, pbar_update
-from util import run_concurrently, tz_aware_datetime
-from util.logging import error_message, warning_message
+from util import fetch_concurrently, run_concurrently, tz_aware_datetime
+from util.logging import warning_message
 
 
 def get_or_create_erratum(name, e_type, issue_date, synopsis):
@@ -105,22 +100,11 @@ def enrich_errata(concurrent_processing=True, max_workers=25):
     pbar_start.send(sender=None, ptext=f'Fetching osv.dev data for {elen} Errata', plen=elen)
     results = []
     if concurrent_processing:
-        session = requests.Session()
-        adapter = HTTPAdapter(pool_connections=max_workers, pool_maxsize=max_workers)
-        session.mount('https://', adapter)
-        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
-            futures = {executor.submit(e.fetch_osv_dev_data, session): e for e in errata}
-            for i, future in enumerate(concurrent.futures.as_completed(futures)):
-                erratum = futures[future]
-                try:
-                    osv_data = future.result()
-                except Exception as e:
-                    error_message(text=f'Error fetching osv.dev data for {erratum}: {e}')
-                    pbar_update.send(sender=None, index=i + 1)
-                    continue
-                if osv_data is not None:
-                    results.append((erratum, osv_data))
-                pbar_update.send(sender=None, index=i + 1)
+        for i, result in enumerate(fetch_concurrently(fetch_osv_worker, errata, max_workers)):
+            erratum, osv_data = result
+            if osv_data is not None:
+                results.append((erratum, osv_data))
+            pbar_update.send(sender=None, index=i + 1)
     else:
         for i, e in enumerate(errata):
             osv_data = e.fetch_osv_dev_data()
@@ -135,3 +119,7 @@ def enrich_errata(concurrent_processing=True, max_workers=25):
         for i, (erratum, osv_data) in enumerate(results):
             erratum.parse_osv_dev_data(osv_data)
             pbar_update.send(sender=None, index=i + 1)
+
+
+def fetch_osv_worker(erratum, session):
+    return (erratum, erratum.fetch_osv_dev_data(session))

util/__init__.py

@@ -300,6 +300,27 @@ def get_datetime_now():
     return datetime.now().astimezone().replace(microsecond=0)
 
 
+def fetch_concurrently(func, items, max_workers=25):
+    """ Run func across items using threads with pooled HTTP sessions,
+        yielding results as they complete. Ideal for I/O-bound work
+        (network fetches). func(item, session) receives a shared
+        requests.Session with connection pooling.
+    """
+    import concurrent.futures
+
+    from requests.adapters import HTTPAdapter
+
+    session = requests.Session()
+    adapter = HTTPAdapter(pool_connections=max_workers, pool_maxsize=max_workers)
+    session.mount('https://', adapter)
+    session.mount('http://', adapter)
+    items = list(items)
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+        futures = {executor.submit(func, item, session): item for item in items}
+        for future in concurrent.futures.as_completed(futures):
+            yield future.result()
+
+
 def run_concurrently(func, items, max_workers=25):
     """ Run func across items using multiprocessing, yielding results as
         they complete. Uses multiprocessing.Pool on Python < 3.12 to avoid