System resource cache

CHANGELOG.rst

@@ -8,6 +8,8 @@ Next release
   ``licensedcode-data``.
   https://github.com/aboutcode-org/scancode-toolkit/pull/5056
 
+- Add experimental option for using cached results during scan time.
+
 v33.0.0rc1 - 2026-05-14
 ------------------------
 

pyproject.toml

@@ -164,7 +164,6 @@ files = [
     { filename = "pyproject.toml" },
     { filename = "pyproject-scancode-toolkit-mini.toml" },
     { filename = "pyproject-packagedcode.toml" },
-
 ]
 
 

src/scancode/cli.py

@@ -67,6 +67,7 @@ class WindowsError(Exception):
 from scancode import notice
 from scancode import print_about
 from scancode import Scanner
+from scancode import results_cache
 from scancode.help import epilog_text
 from scancode.help import examples_text
 from scancode.interrupt import DEFAULT_TIMEOUT
@@ -407,12 +408,19 @@ def default_processes():
     help_group=cliutils.MISC_GROUP, sort_order=1000, cls=PluggableCommandLineOption)
 
 @click.option(
-    "--check-version/--no-check-version",
-    help="Whether to check for new versions. Defaults to true.",
+    '--check-version/--no-check-version',
+    help='Whether to check for new versions. Defaults to true.',
     default=True,
     # not yet supported in Click 6.7 but added in PluggableCommandLineOption
     hidden=True,
     help_group=cliutils.MISC_GROUP, sort_order=1000, cls=PluggableCommandLineOption)
+
+@click.option('--use-cached-results',
+    is_flag=True,
+    default=False,
+    hidden=True,
+    help='(EXPERIMENTAL) ScanCode will use cached results during scan time.',
+    help_group=cliutils.CORE_GROUP, sort_order=250, cls=PluggableCommandLineOption)
 def scancode(
     ctx,
     input,  # NOQA
@@ -433,6 +441,7 @@ def scancode(
     test_error_mode,
     keep_temp_files,
     check_version,
+    use_cached_results,
     echo_func=echo_stderr,
     *args,
     **kwargs,
@@ -549,6 +558,7 @@ def scancode(
             return_results=False,
             echo_func=echo_func,
             outdated=outdated,
+            use_cached_results=use_cached_results,
             *args,
             **kwargs
         )
@@ -570,7 +580,7 @@ def scancode(
 
 
 def run_scan(
-    input,  # 
+    input,  #
     config_file=None,
     ignore=[],
     from_json=False,
@@ -594,6 +604,7 @@ def run_scan(
     pretty_params=None,
     plugin_options=plugin_options,
     outdated=None,
+    use_cached_results=False,
     *args,
     **kwargs
 ):
@@ -1006,6 +1017,7 @@ def echo_func(*_args, **_kwargs):
             verbose=verbose,
             kwargs=requested_options,
             echo_func=echo_func,
+            use_cached_results=use_cached_results,
         )
         success = success and scan_success
 
@@ -1141,7 +1153,7 @@ def load_configuration_file(path):
 
     click.echo(f"Loading env from {path}")
     try:
-        
+
         config_values = saneyaml.load(path.read())
         ignores = config_values.get("ignored_patterns", [])
     except (saneyaml.YAMLError, Exception):
@@ -1226,6 +1238,7 @@ def run_scanners(
     verbose=False,
     kwargs=None,
     echo_func=echo_stderr,
+    use_cached_results=False,
 ):
     """
     Run the list of `stage` ScanPlugin `plugins` on `codebase`.
@@ -1267,7 +1280,8 @@ def run_scanners(
     # TODO: add CLI option to bypass cache entirely?
     scan_success = scan_codebase(
         codebase, scanners, processes, timeout,
-        with_timing=timing, progress_manager=progress_manager)
+        with_timing=timing, progress_manager=progress_manager,
+        use_cached_results=use_cached_results)
 
     # TODO: add progress indicator
     # run the process codebase of each scan plugin (most often a no-op)
@@ -1305,6 +1319,7 @@ def scan_codebase(
     with_timing=False,
     progress_manager=None,
     echo_func=echo_stderr,
+    use_cached_results=False,
 ):
     """
     Run the `scanners` Scanner objects on the `codebase` Codebase. Return True
@@ -1326,15 +1341,16 @@ def scan_codebase(
     """
 
     # NOTE: we never scan directories
-    resources = ((r.location, r.path) for r in codebase.walk() if r.is_file)
+    resources = ((r.location, r.path, r.name) for r in codebase.walk() if r.is_file)
 
     use_threading = processes >= 0
     runner = partial(
         scan_resource,
         scanners=scanners,
         timeout=timeout,
         with_timing=with_timing,
-        with_threading=use_threading
+        with_threading=use_threading,
+        use_cached_results=use_cached_results,
     )
 
     if TRACE:
@@ -1462,11 +1478,12 @@ def terminate_pool_with_backoff(pool, number_of_trials=3):
 
 
 def scan_resource(
-    location_path,
+    location_path_name,
     scanners,
     timeout=DEFAULT_TIMEOUT,
     with_timing=False,
     with_threading=True,
+    use_cached_results=False,
 ):
     """
     Given a ``location_path`` tuple pf (location, path), return a tuple of:
@@ -1489,10 +1506,11 @@ def scan_resource(
     processing and threading works.
     """
     scan_time = time()
-    location, path = location_path
+    location, path, name = location_path_name
     results = {}
     scan_errors = []
     timings = {} if with_timing else None
+    scanners_to_run = []
 
     if not with_threading:
         interruptor = fake_interruptible
@@ -1503,8 +1521,28 @@ def scan_resource(
     # and start returning values. The kill timeout is otherwise there
     # as a gatekeeper for runaway processes.
 
+    results_cache_index = ''
+    if use_cached_results:
+        # compute results_cache_index
+        results_cache_index = results_cache.compute_results_cache_index(location=location, filename=name)
+
+        # update `results` with cached data or add scanner to scanners_to_run if no
+        # cache data is available
+        for scanner in scanners:
+            # get resource_cache_data
+            resource_cache_data = results_cache.get_results_cache_data(
+                results_cache_index=results_cache_index,
+                plugin_name=scanner.name
+            )
+            if resource_cache_data:
+                results.update(resource_cache_data)
+            else:
+                scanners_to_run.append(scanner)
+    else:
+        scanners_to_run = scanners
+
     # run each scanner in sequence in its own interruptible
-    for scanner in scanners:
+    for scanner in scanners_to_run:
         if with_timing:
             start = time()
 
@@ -1523,6 +1561,12 @@ def scan_resource(
             # the return value of a scanner fun MUST be a mapping
             if values_mapping:
                 results.update(values_mapping)
+                if use_cached_results:
+                    results_cache.update_results_cache_data(
+                        results_cache_index=results_cache_index,
+                        plugin_name=scanner.name,
+                        results=values_mapping,
+                    )
 
         except Exception:
             msg = 'ERROR: for scanner: ' + scanner.name + ':\n' + traceback.format_exc()

src/scancode/results_cache.py

@@ -0,0 +1,89 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# ScanCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/nexB/scancode-toolkit for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import hashlib
+import json
+import os
+
+from commoncode.fileutils import create_dir
+from commoncode.hash import binary_chunks
+from scancode_config import results_cache_dir
+
+
+def hasher_from_chunks(chunks):
+    """
+    Return a sha256 hasher loaded with `chunks`.
+    """
+    hasher = hashlib.sha256()
+    for chunk in chunks:
+        hasher.update(chunk)
+    return hasher
+
+
+def compute_results_cache_index(location, filename):
+    """
+    Compute results_cache_index value for a Resource at `location`.
+    """
+    chunks = binary_chunks(location=location)
+    sha256_hasher = hasher_from_chunks(chunks=chunks)
+    sha256_hasher.update(filename.encode('utf-8', 'surrogateescape'))
+    return sha256_hasher.hexdigest()
+
+
+def get_results_cache_directory_location(results_cache_index):
+    """
+    Return the location of the directory containing the cache files for a given
+    `results_cache_index` hexstring.
+    """
+    # Split the hash into two subdirectories using the first two prefix pairs
+    prefix1 = results_cache_index[:2]
+    prefix2 = results_cache_index[2:4]
+    directory_name = results_cache_index[4:]
+    return os.path.join(results_cache_dir, prefix1, prefix2, directory_name)
+
+
+def get_results_cache_file_location(results_cache_index, plugin_name):
+    """
+    Return the location of the file containing the cached results of the scanner
+    `plugin_name` for a resource keyed by `results_cache_index` hexstring.
+    """
+    results_cache_directory_location = get_results_cache_directory_location(results_cache_index=results_cache_index)
+    return os.path.join(results_cache_directory_location, plugin_name)
+
+
+def get_results_cache_data(results_cache_index, plugin_name):
+    """
+    Return a mapping containing the results of scan plugin, `plugin_name`, for a
+    resource keyed by `resource_cache_index` hexstring. If the cache file does
+    not exist, an empty mapping is returned.
+    """
+    results_cache_file_location = get_results_cache_file_location(
+        results_cache_index=results_cache_index,
+        plugin_name=plugin_name
+    )
+    if os.path.exists(results_cache_file_location):
+        with open(results_cache_file_location) as f:
+            return json.load(f)
+    else:
+        return {}
+
+
+def update_results_cache_data(results_cache_index, plugin_name, results):
+    """
+    Update the results cache with the `results` of the scanner `plugin_name`
+    for the resource keyed by `results_cache_index`.
+    """
+    results_cache_directory_location = get_results_cache_directory_location(
+        results_cache_index=results_cache_index
+    )
+    if not os.path.exists(results_cache_directory_location):
+        create_dir(results_cache_directory_location)
+    results_cache_file_location = os.path.join(results_cache_directory_location, plugin_name)
+    with open(results_cache_file_location, 'w') as f:
+        json.dump(results, f)

src/scancode_config.py

@@ -39,8 +39,8 @@
     NOTE: this is essentailly a copy of commoncode.fileutils.create_dir()
     """
 
     if exists(location):
         if not os.path.isdir(location):
             err = ('Cannot create directory: existing file '
                    'in the way ''%(location)s.')
             raise OSError(err % locals())
@@ -49,20 +49,20 @@
     # may fail on win if the path is too long
     # FIXME: consider using UNC ?\\ paths
     try:
         os.makedirs(location)
 
     # avoid multi-process TOCTOU conditions when creating dirs
     # the directory may have been created since the exist check
     except WindowsError as e:
         # [Error 183] Cannot create a file when that file already exists
         if e and e.winerror == 183:
             if not os.path.isdir(location):
                 raise
         else:
             raise
     except (IOError, OSError) as o:
         if o.errno == errno.EEXIST:
             if not os.path.isdir(location):
                 raise
         else:
             raise
@@ -190,9 +190,14 @@
 __env_package_cache_dir = os.getenv('SCANCODE_PACKAGE_INDEX_CACHE')
 packagedcode_cache_dir = (__env_package_cache_dir or std_package_cache_dir)
 
+std_results_cache_dir = join(scancode_cache_dir, 'results')
+__env_results_cache_dir = os.getenv('SCANCODE_RESULTS_CACHE')
+results_cache_dir = (__env_results_cache_dir or std_results_cache_dir)
+
 _create_dir(licensedcode_cache_dir)
 _create_dir(packagedcode_cache_dir)
 _create_dir(scancode_cache_dir)
+_create_dir(results_cache_dir)
 
 # - scancode_temp_dir: for short-lived temporary files which are import- or run-
 # specific that may live for the duration of a function call or for the duration