async def METHOD_NAME(
    request: Flake8Request.Batch[Flake8FieldSet, InterpreterConstraints],
    flake8: Flake8,
    first_party_plugins: Flake8FirstPartyPlugins,
) -> LintResult:
    interpreter_constraints = request.partition_metadata
    flake8_pex_get = Get(
        VenvPex,
        PexRequest,
        flake8.to_pex_request(
            interpreter_constraints=interpreter_constraints,
            extra_requirements=first_party_plugins.requirement_strings,
        ),
    )
    config_files_get = Get(ConfigFiles, ConfigFilesRequest, flake8.config_request)
    source_files_get = Get(
        SourceFiles, SourceFilesRequest(field_set.source for field_set in request.elements)
    )
    extra_files_get = Get(
        Digest,
        PathGlobs(
            globs=flake8.extra_files,
            glob_match_error_behavior=GlobMatchErrorBehavior.error,
            description_of_origin="the option [flake8].extra_files",
        ),
    )
    # Ensure that the empty report dir exists.
    report_directory_digest_get = Get(Digest, CreateDigest([Directory(REPORT_DIR)]))
    flake8_pex, config_files, report_directory, source_files, extra_files = await MultiGet(
        flake8_pex_get,
        config_files_get,
        report_directory_digest_get,
        source_files_get,
        extra_files_get,
    )
    input_digest = await Get(
        Digest,
        MergeDigests(
            (
                source_files.snapshot.digest,
                first_party_plugins.sources_digest,
                config_files.snapshot.digest,
                extra_files,
                report_directory,
            )
        ),
    )
    result = await Get(
        FallibleProcessResult,
        VenvPexProcess(
            flake8_pex,
            argv=generate_argv(source_files, flake8),
            input_digest=input_digest,
            output_directories=(REPORT_DIR,),
            extra_env={"PEX_EXTRA_SYS_PATH": first_party_plugins.PREFIX},
            concurrency_available=len(request.elements),
            description=f"Run Flake8 on {pluralize(len(request.elements), 'file')}.",
            level=LogLevel.DEBUG,
        ),
    )
    report = await Get(Digest, RemovePrefix(result.output_digest, REPORT_DIR))
    return LintResult.create(request, result, report=report)