Function generate_api_spec_via_fmt

pants-plugins/api_spec/rules.py:67–111

(
    request: GenerateAPISpecViaFmtTargetsRequest.Batch,
    subsystem: GenerateApiSpec,
)

    level=LogLevel.DEBUG,
)
async def generate_api_spec_via_fmt(
    request: GenerateAPISpecViaFmtTargetsRequest.Batch,
    subsystem: GenerateApiSpec,
) -> FmtResult:
    config_files_get = Get(ConfigFiles, ConfigFilesRequest, subsystem.config_request())

    # We use a pex to actually generate the api spec with an external script.
    # Generation cannot be inlined here because it needs to import the st2 code.
    # (the script location is defined on the GenerateApiSpec subsystem)
    pex_get = Get(VenvPex, PexFromTargetsRequest, subsystem.pex_request())

    config_files, pex = await MultiGet(config_files_get, pex_get)

    input_digest = await Get(
        Digest,
        MergeDigests((config_files.snapshot.digest, request.snapshot.digest)),
    )

    result = await Get(
        ProcessResult,
        VenvPexProcess(
            pex,
            argv=(
                "--config-file",
                subsystem.config_file,
            ),
            input_digest=input_digest,
            description="Regenerating openapi.yaml api spec",
            level=LogLevel.DEBUG,
        ),
    )

    contents = [FileContent(file, result.stdout) for file in request.files]

    output_digest = await Get(Digest, CreateDigest(contents))
    output_snapshot = await Get(Snapshot, Digest, output_digest)

    return FmtResult(
        input=request.snapshot,
        output=output_snapshot,
        # Drop result.stdout since we already wrote it to a file
        stdout="",
        stderr=strip_v2_chroot_path(result.stderr),
        tool_name=request.tool_name,
    )


@rule(