Function scan_codebase

src/scancode/cli.py:1254–1393

Run the `scanners` Scanner objects on the `codebase` Codebase. Return True on success or False otherwise. Use multiprocessing with `processes` number of processes defaulting to single process. Disable multiprocessing with processes 0 or -1. Disable threading is processes is -1.

(
    codebase,
    scanners,
    processes=1,
    timeout=DEFAULT_TIMEOUT,
    with_timing=False,
    progress_manager=None,
    echo_func=echo_stderr,
)

def scan_codebase(
    codebase,
    scanners,
    processes=1,
    timeout=DEFAULT_TIMEOUT,
    with_timing=False,
    progress_manager=None,
    echo_func=echo_stderr,
):
    """
    Run the `scanners` Scanner objects on the `codebase` Codebase. Return True
    on success or False otherwise.

    Use multiprocessing with `processes` number of processes defaulting to
    single process. Disable multiprocessing with processes 0 or -1. Disable
    threading is processes is -1.

    Run each scanner function for up to `timeout` seconds and fail it otherwise.

    If `with_timing` is True, each Resource is updated with per-scanner
    execution time (as a float in seconds). This is added to the `scan_timings`
    mapping of each Resource as {scanner.name: execution time}.

    Provide optional progress feedback in the UI using the ``progress_manager``
    callable that accepts an iterable of tuple of (location, path, scan_errors,
    scan_result) as argument.
    """

    # NOTE: we never scan directories
    resources = ((r.location, r.path) 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
    )

    if TRACE:
        logger_debug('scan_codebase: scanners:', ', '.join(s.name for s in scanners))

    get_resource = codebase.get_resource

    success = True
    pool = None
    scans = None
    try:
        if processes >= 1:
            # maxtasksperchild helps with recycling processes in case of leaks
            pool = get_pool(processes=processes, maxtasksperchild=1000)
            # Using chunksize is documented as much more efficient in the Python
            # doc. Yet "1" still provides a better and more progressive
            # feedback. With imap_unordered, results are returned as soon as
            # ready and out of order so we never know exactly what is processing
            # until completed.
            scans = pool.imap_unordered(runner, resources, chunksize=1)