Use `uv`’s

--extra

support if your project already depends on

prefect

, or add the dependency explicitly. Install into the current environment/project:

bash
uv add 'prefect[aws]'

Or with pip-style invocation via uv:

bash
uv pip install 'prefect[aws]'

If you just want to run something with the extra available without modifying dependencies:

bash
uv run --with 'prefect[aws]' python your_script.py

If helpful, I can also show the exact

pyproject.toml

entry or how to verify the AWS integration imported correctly.

For Prefect 3.x, both are valid: - create blocks from code with

.save()

- create/manage them from the UI The best choice depends on what you want to optimize for. Typical recommendation - Create block types/instances from code if you want reproducibility and setup to live alongside your infrastructure/app code - Use the UI if you want easier manual management of credentials/config after initial setup For S3 specifically, the usual pattern is: 1. Create an

AwsCredentials

block 2. Create an

S3Bucket

block that references those credentials 3. Load the

S3Bucket

block inside your flow/task and use it to read/write objects Example:

python
from prefect_aws import AwsCredentials
from prefect_aws.s3 import S3Bucket

creds = AwsCredentials(
    aws_access_key_id="...",
    aws_secret_access_key="...",
    region_name="us-east-1",
)
creds.save("my-aws-creds", overwrite=True)

bucket = S3Bucket(
    bucket_name="my-bucket",
    credentials=creds,
    bucket_folder="my/prefix",  # optional
)
bucket.save("my-s3-bucket", overwrite=True)

Then in a flow:

python
from prefect import flow
from prefect_aws.s3 import S3Bucket

@flow
def my_flow():
    bucket = S3Bucket.load("my-s3-bucket")

    bucket.write_path("example.txt", b"hello")
    data = bucket.read_path("example.txt")
    print(data)

Should I create them in code or in the UI? Use code when: - you want repeatable environment setup - you manage infra as code - you want dev/staging/prod bootstrapping scripts - you want block creation to be version-controlled Use the UI when: - an operator/admin manages secrets manually - credentials rotate outside your app code - you want a quick way to inspect/update config A very common hybrid approach is: - define and bootstrap blocks once in code - allow updates in the UI later if needed Security note For AWS secrets, storing them in an

AwsCredentials

block is fine; sensitive fields are treated as secrets and stored encrypted by Prefect. But if your runtime already has AWS auth via environment variables or IAM role, you may not need to hardcode keys into a block at all. For example, if your worker runs with an IAM role, you can often do:

python
from prefect_aws.s3 import S3Bucket

bucket = S3Bucket(bucket_name="my-bucket")
bucket.save("my-s3-bucket", overwrite=True)

Then boto3/Python AWS auth resolution can pick up credentials from the environment/role at runtime. Important Prefect 3.x note Blocks are still supported and useful, but for Git-based code storage you should not use old GitHub storage blocks from Prefect 2.x. For source code, use

.from_source(...)

instead. For S3 object access, though,

S3Bucket

blocks are still a good fit. Useful docs: - Prefect blocks concepts - `prefect_aws.s3` API reference - `prefect_aws` credentials reference If you want, I can also give you: - a full example flow uploading/downloading a file from S3, or - the recommended pattern for using S3 blocks as result storage/cache storage.

Here’s a recommended pattern that works well in Prefect 3.x for S3 object storage with

prefect-aws

. Recommendation in short - Create reusable blocks once in a bootstrap/setup step - Load the

S3Bucket

block inside flows/tasks - Prefer IAM roles or environment-provided AWS credentials in production instead of hardcoding access keys - Use

AwsCredentials

+

S3Bucket

blocks when you want centralized, reusable config in Prefect --- ## 1) Install With `uv`:

bash
uv add 'prefect[aws]'

Or for one-off execution:

bash
uv run --with 'prefect[aws]' python your_script.py

--- ## 2) Recommended setup pattern There are really two good production patterns. ### Pattern A: Best for AWS-hosted workloads If your worker runs on EC2, ECS, EKS, Lambda, etc., and already has an IAM role: - don’t store access keys in the block - just store region/config and bucket metadata - let boto3 resolve credentials from the environment/role ### Pattern B: Best for centralized Prefect-managed config If you need explicit credentials managed in Prefect: - create an

AwsCredentials

block - create an

S3Bucket

block that references it - load the bucket block in flows --- ## 3) Bootstrap script: create blocks once This script creates: - an

AwsCredentials

block - an

S3Bucket

block You’d run it once per environment.

python
import os

from prefect_aws import AwsCredentials
from prefect_aws.s3 import S3Bucket


def bootstrap_blocks():
    creds = AwsCredentials(
        aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
        aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
        region_name=os.getenv("AWS_DEFAULT_REGION", "us-east-1"),
    )
    creds.save("prod-aws-creds", overwrite=True)

    bucket = S3Bucket(
        bucket_name="my-data-bucket",
        credentials=creds,
        bucket_folder="prefect/data",
    )
    bucket.save("prod-s3-bucket", overwrite=True)

    print("Saved blocks: prod-aws-creds, prod-s3-bucket")


if __name__ == "__main__":
    bootstrap_blocks()

If you’re using IAM roles instead of keys, use this version instead:

python
from prefect_aws import AwsCredentials
from prefect_aws.s3 import S3Bucket


def bootstrap_blocks():
    creds = AwsCredentials(
        region_name="us-east-1",
    )
    creds.save("prod-aws-creds", overwrite=True)

    bucket = S3Bucket(
        bucket_name="my-data-bucket",
        credentials=creds,
        bucket_folder="prefect/data",
    )
    bucket.save("prod-s3-bucket", overwrite=True)

    print("Saved blocks: prod-aws-creds, prod-s3-bucket")


if __name__ == "__main__":
    bootstrap_blocks()

In that case, boto3 will use the worker’s IAM role / env vars / AWS config chain at runtime. --- ## 4) Full flow example: upload, read back, list, and download to local disk ```python import json from pathlib import Path from prefect import flow, task, get_run_logger from prefect_aws.s3 import S3Bucket @task def build_payload() -> bytes: payload = { "records": [ {"id": 1, "name": "alice"}, {"id": 2, "name": "bob"}, ] } return json.dumps(payload).encode() @task def upload_bytes(bucket: S3Bucket, key: str, content: bytes) -> str: logger = get_run_logger() written_key = bucket.write_path(key, content) logger.info(f"Uploaded object to s3://{bucket.bucket_name}/{written_key}") return written_key @task def read_bytes(bucket: S3Bucket, key: str) -> dict: logger = get_run_logger() raw = bucket.read_path(key) logger.info(f"Read object from s3://{bucket.bucket_name}/{key}") return json.loads(raw.decode()) @task def list_prefix(bucket: S3Bucket, prefix: str): logger = get_run_logger() objects = bucket.list_objects(folder=prefix) logger.info(f"Found {len(objects)} objects under prefix '{prefix}'") return objects @task def upload_local_file(bucket: S3Bucket, local_path: str, remote_key: str) -> str: logger = get_run_logger()

uploaded_key = bucket.upload_from_path(from_path=local_path, to_path=remote_key) logger.info(f"Uploaded file to s3://{bucket.bucket_name}/{uploaded_key}") return uploaded_key @task def download_to_local(bucket: S3Bucket, remote_key: str, local_path: str) -> str: logger = get_run_logger() result = bucket.download_object_to_path(from_path=remote_key, to_path=local_path) logger.info(f"Downloaded s3://{bucket.bucket_name}/{remote_key} to {result}") return str(result) @flow def s3_block_example(): logger = get_run_logger() bucket = S3Bucket.load("prod-s3-bucket") data_key = "runs/example/data.json" content = build_payload() uploaded_key = upload_bytes(bucket, data_key, content) parsed = read_bytes(bucket, uploaded_key) logger.info(f"Parsed payload: {parsed}") local_file = Path("sample.txt") local_file.write_text("hello from prefect\n") uploaded_file_key = upload_local_file( bucket, local_path=str(local_file), remote_key="runs/example/sample.txt", ) download_to_local( bucket, remote_key=uploaded_file_key, local_path="downloaded-sample.txt", ) objects = list_prefix(bucket, "runs/example/") for obj in objects: logger.info(f"Object: {obj['Key']} ({obj['Size']} bytes)") return { "json_key": uploaded_key, "file_key": uploaded_file_key, "object_count": len(objects), } if name == "__main__": s3_block_example()

---

## 5) Why this is the recommended pattern

### *Good separation of concerns*
- bootstrap code manages infrastructure/config
- flow code just loads and uses the block

### *Reusable across flows*
Any flow can do:

```python
bucket = S3Bucket.load("prod-s3-bucket")

and immediately use the same configuration. ### Easier credential rotation If credentials change, you update the block once instead of changing many flows. ### Cleaner deployment code Your flows don’t need to know where AWS keys come from. --- ## 6) Recommended production variants ### Variant 1: IAM role preferred If your worker runs in AWS, this is usually the best pattern. Bootstrap:

python
from prefect_aws import AwsCredentials
from prefect_aws.s3 import S3Bucket

creds = AwsCredentials(region_name="us-east-1")
creds.save("prod-aws-creds", overwrite=True)

bucket = S3Bucket(
    bucket_name="my-data-bucket",
    credentials=creds,
    bucket_folder="prefect/data",
)
bucket.save("prod-s3-bucket", overwrite=True)

Why this is best: - no long-lived keys stored in code - simpler secret management - aligns with AWS best practices ### Variant 2: explicit credentials in Prefect Use when IAM roles aren’t practical. Bootstrap from environment variables, not hardcoded literals:

python
import os
from prefect_aws import AwsCredentials

creds = AwsCredentials(
    aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
    aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
    region_name=os.getenv("AWS_DEFAULT_REGION", "us-east-1"),
)
creds.save("prod-aws-creds", overwrite=True)

That keeps the actual secrets outside source control. --- ## 7) Common operations you’ll probably use ### Read object bytes

python
bucket = S3Bucket.load("prod-s3-bucket")
data = bucket.read_path("folder/file.json")

### Write object bytes

python
bucket.write_path("folder/file.json", b'{"ok": true}')

### Upload a local file

python
bucket.upload_from_path(from_path="local.csv", to_path="uploads/local.csv")

### Download to local path

python
bucket.download_object_to_path(
    from_path="uploads/local.csv",
    to_path="downloaded.csv",
)

### List objects

python
objects = bucket.list_objects(folder="uploads/")

### Upload a whole folder

python
bucket.upload_from_folder(from_folder="data", to_folder="archive")

--- ## 8) UI vs code: what I’d actually do My practical recommendation:

- Use code for initial creation in dev/staging/prod - Use the UI for inspection or occasional updates - keep the bootstrap script in version control That gives you reproducibility and operational convenience. --- ## 9) What not to do ### Avoid embedding raw credentials directly in flow code Not ideal:

python
@flow
def my_flow():
    ...

with access keys hardcoded inline. ### Avoid re-creating blocks every flow run Blocks are meant to be saved once and loaded many times. ### Avoid old Prefect 2.x storage guidance For Prefect 3.x, S3 blocks are fine for object storage, but don’t use deprecated 2.x GitHub storage block patterns for code storage. --- ## 10) Simple “best practice” template If you want the shortest version of the recommended approach, use this: ### Setup once

python
from prefect_aws import AwsCredentials
from prefect_aws.s3 import S3Bucket

creds = AwsCredentials(region_name="us-east-1")
creds.save("aws-creds", overwrite=True)

bucket = S3Bucket(
    bucket_name="my-bucket",
    credentials=creds,
    bucket_folder="project-a",
)
bucket.save("project-a-bucket", overwrite=True)

### Use in flows

python
from prefect import flow
from prefect_aws.s3 import S3Bucket

@flow
def my_flow():
    bucket = S3Bucket.load("project-a-bucket")
    bucket.write_path("output.txt", b"hello")
    return bucket.read_path("output.txt")

--- ## Docs - Blocks concepts - `prefect_aws.s3` API reference - `prefect_aws` credentials reference If you want, I can also give you a deployment-ready example with: -

prefect.yaml

-

uv run --with 'prefect[aws]'

- worker/job configuration - result storage on S3

Yes — you can bootstrap Global Concurrency Limits in Python in Prefect 3.x using the Prefect client. Recommended pattern: use an idempotent bootstrap script that runs during environment setup/deploy and uses

upsert_global_concurrency_limit_by_name(...)

. --- ## Full Python example ### Async version

python
import asyncio

from prefect.client.orchestration import get_client


LIMITS = [
    {"name": "database", "limit": 5},
    {"name": "openai-api", "limit": 20, "slot_decay_per_second": 1.0},
    {"name": "snowflake", "limit": 10},
]


async def bootstrap_gcls():
    async with get_client() as client:
        for item in LIMITS:
            await client.upsert_global_concurrency_limit_by_name(
                name=item["name"],
                limit=item["limit"],
                slot_decay_per_second=item.get("slot_decay_per_second"),
            )
            print(f"Upserted GCL: {item['name']} -> {item['limit']}")


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

This is the cleanest option because it is safe to rerun. --- ## Sync version

python
from prefect.client.orchestration import get_client


def bootstrap_gcls():
    with get_client(sync_client=True) as client:
        client.upsert_global_concurrency_limit_by_name(
            name="database",
            limit=5,
        )

        client.upsert_global_concurrency_limit_by_name(
            name="openai-api",
            limit=20,
            slot_decay_per_second=1.0,
        )

        print("Bootstrapped GCLs")


if __name__ == "__main__":
    bootstrap_gcls()

--- ## If you want explicit create/update APIs You can also use the lower-level create/update models.

python
import asyncio

from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import (
    GlobalConcurrencyLimitCreate,
    GlobalConcurrencyLimitUpdate,
)


async def manage_gcl():
    async with get_client() as client:
        gcl_id = await client.create_global_concurrency_limit(
            GlobalConcurrencyLimitCreate(
                name="my-limit",
                limit=10,
                active=True,
                slot_decay_per_second=0.0,
            )
        )
        print(f"Created: {gcl_id}")

        limit = await client.read_global_concurrency_limit_by_name("my-limit")
        print(limit)

        await client.update_global_concurrency_limit(
            "my-limit",
            GlobalConcurrencyLimitUpdate(limit=25, active=True),
        )

        limits = await client.read_global_concurrency_limits(limit=100, offset=0)
        print([l.name for l in limits])

        await client.delete_global_concurrency_limit_by_name("my-limit")

--- ## Recommended bootstrap pattern I’d suggest: 1. Keep desired limits in a list/dict 2. Run a bootstrap script in CI/CD or environment setup 3. Use

upsert_global_concurrency_limit_by_name(...)

4. Treat limits as code, like infra config Example:

python
from prefect.client.orchestration import get_client

DESIRED_GCLS = {
    "database": {"limit": 5},
    "salesforce-api": {"limit": 10, "slot_decay_per_second": 0.5},
    "bigquery": {"limit": 8},
}


def bootstrap():
    with get_client(sync_client=True) as client:
        for name, config in DESIRED_GCLS.items():
            client.upsert_global_concurrency_limit_by_name(
                name=name,
                limit=config["limit"],
                slot_decay_per_second=config.get("slot_decay_per_second"),
            )
            print(f"Ensured {name}")


if __name__ == "__main__":
    bootstrap()

--- ## Concurrency vs rate limiting

slot_decay_per_second

changes behavior: -

0.0

or omitted: acts like a standard concurrency cap - `> 0`: acts more like a rate limit, where slots decay over time Examples: - DB connection pool:

limit=5

, no decay - External API:

limit=100

,

slot_decay_per_second=1.0

--- ## Reading and validating existing limits ```python

from prefect.client.orchestration import get_client def show_limits(): with get_client(sync_client=True) as client: limits = client.read_global_concurrency_limits(limit=100, offset=0) for gcl in limits: print( gcl.name, gcl.limit, gcl.active, gcl.active_slots, gcl.slot_decay_per_second, )

---

## CLI exists too

Verified CLI:

```bash
prefect global-concurrency-limit --help

Available commands include: -

create

-

ls

-

inspect

-

update

-

enable

-

disable

-

delete

If you want CLI docs too, I can provide exact commands. --- ## Docs / references - Global Concurrency Limits - Prefect client reference If you want, I can also show: - how to apply a GCL in a flow/task, or - a full “bootstrap + use limit in tasks” example.