Skip to main content

OutputStream

The output stream for bxl to print values to the console as their result

OutputStream.ensure​

def OutputStream.ensure(
artifact: artifact,
) -> bxl.EnsuredArtifact

Marks the artifact as an artifact that should be available to the users at the end of the bxl invocation. Any artifacts that do not get registered via this call is not accessible by users at the end of bxl script.

This function returns an ensured_artifact type that can be printed via ctx.output.print() to print its actual path on disk.

Sample usage:

def _impl_ensure(ctx):
    actions = ctx.bxl_actions().actions
    output = actions.write("my_output", "my_content")
    ensured = ctx.output.ensure(output)
    ctx.output.print(ensured)

OutputStream.ensure_multiple​

def OutputStream.ensure_multiple(
artifacts: CellPath | CellRoot | None | OutputArtifact | ProjectRoot | ResolvedStringWithMacros | TaggedCommandLine | TargetLabel | TransitiveSetArgsProjection | WriteJsonCliArgs | artifact | bxl.BuildResult | bxl.BuiltArtifactsIterable | cmd_args | label | str | list[artifact] | dict[typing.Any, bxl.BuildResult] | RunInfo,
)

Same as ensure, but for multiple artifacts. Will preserve the shape of the inputs (i.e. if the resulting Dict of a ctx.build() is passed in, the output will be a Dict where the key is preserved, and the values are converted to ensured_artifacts).

Note that is slower to loop through objects and ensure them one by one than it is to call ensure_multiple() on all the objects at once (if possible). So, it is suggested to use this method when you are only ensuring a few individual artifacts that are not stored in an iterable.

Sample usage:

def _impl_ensure_multiple(ctx):
    outputs = {}
    for target, value in ctx.build(ctx.cli_args.target).items():
    outputs.update({target.raw_target(): ctx.output.ensure_multiple(value.artifacts())})
    ctx.output.print_json(outputs)

OutputStream.print​

def OutputStream.print(
*args,
sep: str = " ",
) -> None

Outputs results to the console via stdout. These outputs are considered to be the results of a bxl script, which will be displayed to stdout by buck2 even when the script is cached. Accepts an optional separator that defaults to " ".

Prints that are not result of the bxl should be printed via stderr via the stdlib print and pprint.

Note that ctx.output.print() is intended for simple outputs. For more complex outputs, the recommendation would be to write them to a file.

Note that the output of ctx.output.print() will be displayed to stdout at the end of bxl. If you want to print something to stdout immediately or the when ensured artifacts are ready and also want this even when the script is cached, use ctx.output.stream() instead.

Sample usage:

def _impl_print(ctx):
    ctx.output.print("test")

OutputStream.print_json​

def OutputStream.print_json(
value,
*,
pretty: bool = True,
) -> None

Outputs results to the console via stdout as pretty-printed json. Pretty printing can be turned off by the pretty keyword-only parameter. These outputs are considered to be the results of a bxl script, which will be displayed to stdout by buck2 even when the script is cached.

Prints that are not result of the bxl should be printed via stderr via the stdlib print and pprint.

Note that the output of ctx.output.print_json() will be displayed to stdout at the end of bxl. If you want to print something to stdout immediately or the when ensured artifacts are ready and also want this even when the script is cached, use ctx.output.stream_json() instead.

Sample usage:

def _impl_print_json(ctx):
    outputs = {}
    outputs.update({"foo": bar})
    ctx.output.print_json(outputs)

OutputStream.stream​

def OutputStream.stream(
*args,
sep: str = " ",
wait_on: list[bxl.EnsuredArtifact | bxl.EnsuredArtifactGroup] = [],
) -> None

Streaming outputs results to the console via stdout immediately when it is ready. It will be displayed to stdout by buck2 even when the script is cached. Accepts an optional separator that defaults to " ".

The streaming behavior is controlled by the wait_on parameter:

  • If 'wait_on' is not specified or empty, output is displayed immediately during evaluation
  • If 'wait_on' contains artifacts，output is displayed as soon as all specified artifacts are materialized

The wait_on parameter explicitly specifies artifacts that must be materialized before showing output.

Sample usage:

def _impl_print(ctx):
    # Immediate output during evaluation
    ctx.output.stream("Starting process...")

    # Output as soon as artifact is materialized
    artifact = ctx.output.ensure(my_artifact)
    ctx.output.stream("Artifact ready:", artifact, wait_on=[artifact])

    # Output when both artifacts are materialized
    artifact1 = ctx.output.ensure(my_artifact1)
    artifact2 = ctx.output.ensure(my_artifact2)
    ctx.output.stream("First artifact:", artifact1, wait_on=[artifact1, artifact2])

OutputStream.stream_json​

def OutputStream.stream_json(
value,
*,
pretty: bool = True,
wait_on: list[bxl.EnsuredArtifact | bxl.EnsuredArtifactGroup] = [],
) -> None

Streaming outputs results to the console via stdout as pretty-printed json immediately when it is ready. Pretty printing can be turned off by the pretty keyword-only parameter. It will be displayed to stdout by buck2 even when the script is cached.

The streaming behavior is controlled by the wait_on parameter:

  • If 'wait_on' is not specified or empty, output is displayed immediately during evaluation
  • If 'wait_on' contains artifacts，output is displayed as soon as all specified artifacts are materialized

The wait_on parameter explicitly specifies artifacts that must be materialized before showing output.

Sample usage:

def _impl_print_json(ctx):
    outputs = {}
    outputs.update({"foo": bar})

    # Stream JSON output immediately
    ctx.output.stream_json({"status": "starting"})

    # Stream JSON when artifact is ready
    artifact = ctx.output.ensure(my_artifact)
    ctx.output.stream_json({"artifact": artifact}, wait_on=[artifact])

   # Stream JSON waiting on artifact to be ready
   ctx.output.stream_json({"status": "starting"}, wait_on=[artifact])