如何在Prefect中正确使用Blocks加载AWS S3凭证？

问题原因 #

你遇到的AttributeError是因为AwsCredentials.load()是异步方法，直接调用不会返回凭证实例，而是返回一个coroutine对象，所以无论是直接传给s3_download还是访问aws_access_key_id属性都会报错。

正确使用方式 #

在异步流函数中，必须用await关键字调用AwsCredentials.load()，获取实际的AWS凭证实例后再传入s3_download。

完整可运行代码 #

import asyncio

from prefect_aws.s3 import s3_download
from prefect_aws.credentials import AwsCredentials

from prefect import flow, get_run_logger


@flow
async def fetch_taxi_data():
    logger = get_run_logger()
    # 用await加载AWS Credentials Block
    aws_credentials_block = await AwsCredentials.load("aws-credentials-block")
    data = await s3_download(
        bucket="hongbomiao-bucket",
        key="hm-airflow/taxi.csv",
        aws_credentials=aws_credentials_block,
    )
    logger.info(data)

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

补充说明 #

• 完全推荐使用Prefect Blocks管理凭证，这是官方最佳实践，能避免硬编码敏感信息，还能统一管理、动态更新凭证。
• 所有Prefect Block的load()方法都是异步的，在异步上下文中必须用await调用；如果是同步函数，需要用asyncio.run()包裹调用。

内容的提问来源于stack exchange，提问作者Hongbo Miao