experimental prefect-aws bundle steps

[zzstoatzz]

this PR adds the code for transporting bundles to and from s3, allowing execution of the bundle directly from s3 with a Runner

• upload bundle to s3
• download and execute bundle

example

import asyncio
import subprocess
from pathlib import Path
from typing import Any

from pydantic_core import to_json

from prefect import flow, get_client
from prefect._experimental.bundles import create_bundle_for_flow_run

BUCKET_NAME = "zestoatbucket"


async def _write_sample_bundle(parameters: dict[str, Any]):
    @flow(log_prints=True)
    def add_one(x: int) -> int:
        _sum = x + 1
        print(f"x: {x}, _sum: {_sum}")
        return _sum

    async with get_client() as client:
        flow_run = await client.create_flow_run(
            flow=add_one,
            parameters=parameters,
        )

    bundle = create_bundle_for_flow_run(add_one, flow_run)

    Path("foo").write_bytes(to_json(bundle))


async def main():
    ### WORKER SUBMIT STUFF

    await _write_sample_bundle(parameters={"x": 42})

    subprocess.run(
        [
            "uv",
            "run",
            "-m",
            "prefect_aws.experimental.bundles.upload",
            "foo",
            "--bucket",
            BUCKET_NAME,
            "--key",
            "foo",
        ]
    )

    ### ENTRYPOINT FOR EVENTUAL RUNTIME

    subprocess.run(
        [
            "uv",
            "run",
            "-m",
            "prefect_aws.experimental.bundles.execute",
            "--bucket",
            BUCKET_NAME,
            "--key",
            "foo",
        ]
    )


if __name__ == "__main__":
    asyncio.run(main())

results in

14:12:32.972 | INFO    | Flow run 'outgoing-bobcat' - Beginning flow run 'outgoing-bobcat' for flow 'add-one'
14:12:32.974 | INFO    | Flow run 'outgoing-bobcat' - View at http://localhost:4200/runs/flow-run/461d15c2-4e9c-4af6-a313-07c5b7c28d8c
14:12:32.975 | INFO    | Flow run 'outgoing-bobcat' - x: 42, _sum: 43
14:12:32.987 | INFO    | Flow run 'outgoing-bobcat' - Finished in state Completed()
14:12:33.076 | INFO    | prefect.flow_runs.runner - Process for flow run 'outgoing-bobcat' exited cleanly.

Related to #17194

[codspeed-hq]

CodSpeed Performance Report

Merging #17201 will not alter performance

_{Comparing bundle-steps (516919a) with main (8e802bb)}

Summary

✅ 2 untouched benchmarks

[desertaxle]

Implementation looks good! Could you add some tests for the CLI interface also? We'll want to ensure that behavior is maintained.

[cicdw]

cool use of typer - left some minor bookkeeping comments, otherwise this makes sense to me

[cicdw]

two notes:

• worth calling out if block is None, credentials inferred from environment
• prob worth doc'ing the return value (what is generally in the dict?)

[cicdw]

given the reliance on the local_path key, could be worth elevating the return type of download_bundle to a TypedDict or dataclass

[cicdw]

same super minor point: could be worth a structured return object

[desertaxle]

One small question on test assertions, but otherwise LGTM!

[desertaxle]

Could we verify these values get passed into the function as expected?