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

【问题标题】：How to use Blocks correctly to load AWS S3 credentials in Prefect?如何正确使用 Blocks 在 Prefect 中加载 AWS S3 凭证？
【发布时间】：2022-09-27 16:09:53
【问题描述】：

我正在使用 Prefect。我试图从 S3 下载一个文件。

当我硬编码 AWS 凭证时，可以成功下载文件：

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()
    credentials = AwsCredentials(
        aws_access_key_id=\"xxx\",
        aws_secret_access_key=\"xxx\",
    )
    data = await s3_download(
        bucket=\"hongbomiao-bucket\",
        key=\"hm-airflow/taxi.csv\",
        aws_credentials=credentials,
    )
    logger.info(data)

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

现在我尝试从Prefect Blocks 加载凭据。

我创建了一个 AWS 凭证块：

然而，

aws_credentials_block = AwsCredentials.load(\"aws-credentials-block\")
data = await s3_download(
    bucket=\"hongbomiao-bucket\",
    key=\"hm-airflow/taxi.csv\",
    aws_credentials=aws_credentials_block,
)

抛出错误：

AttributeError: \'coroutine\' 对象没有属性 \'get_boto3_session\'

和
aws_credentials_block = AwsCredentials.load(\"aws-credentials-block\")
credentials = AwsCredentials(
    aws_access_key_id=aws_credentials_block.aws_access_key_id,
    aws_secret_access_key=aws_credentials_block.aws_secret_access_key,
)
data = await s3_download(
    bucket=\"hongbomiao-bucket\",
    key=\"hm-airflow/taxi.csv\",
    aws_credentials=credentials,
)
抛出错误：

AttributeError: \'coroutine\' 对象没有属性 \'aws_access_key_id\'

我没有找到任何关于如何使用它的有用文档。

我应该使用 Blocks 来加载凭据吗？如果是，在 Prefect 中正确使用 Blocks 的正确方法是什么？谢谢！

标签： python prefect

【解决方案1】：

我刚刚发现问题截图中的官方演示具有误导性。它错过了await。

添加await后，就可以使用了！

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,
)

我会开一张票报告这个问题。

【讨论】：