How to use PrefectSecret with (Azure) Task Library and what are the differences between secrets in a module vs. Flow scope?

Archive
, , , , , ,
1

From this, Secrets | Prefect Docs

with Flow("example") as flow:
    my_secret = PrefectSecret("MYSECRET")
    res = my_task(credentials=my_secret)

I expected this to work:

from prefect import Flow
from prefect.tasks.secrets import PrefectSecret
from prefect.tasks.azure.blobstorage import BlobStorageDownload

secret = PrefectSecret("azure")

task = BlobStorageDownload(
        azure_credentials_secret=secret,
        container="prefect"
)
task.run(blob_name="prefect.txt")

Resulting in this error:
ValueError: Local Secret "<Task: azure>" was not found.

But actually, I needed to get rid of PrefectSecret

from prefect import Flow
from prefect.tasks.secrets import PrefectSecret
from prefect.tasks.azure.blobstorage import BlobStorageDownload

secret = "azure"

task = BlobStorageDownload(
        azure_credentials_secret=secret,
        container="prefect"
)
task.run(blob_name="prefect.txt")

Should PrefectSecret work with azure_credentials_secret?

2

Good question. First, you can use this setting to ensure you are using Secrets from Cloud backend rather than local secrets:

export PREFECT__CLOUD__USE_LOCAL_SECRETS=false

Then, calling PrefectSecret doesn’t work on the module level, only within the Flow block since this task retrieves secrets at runtime, and module scope gets interpreted at build time (e.g. when you register or visualize your flow). So you would usually need to do that instead:

from prefect import Flow
from prefect.tasks.secrets import PrefectSecret
from prefect.tasks.azure.blobstorage import BlobStorageDownload


azure_task = BlobStorageDownload(container="prefect")

with Flow("example") as flow:
    my_secret = PrefectSecret("azure")
    res = azure_task(azure_credentials_secret=my_secret)

But when you look at the docs of the BlobStorageDownload task Azure Tasks | Prefect Docs you’ll see that it expects the name of the Secret, rather than the secret itself. That’s why you don’t need to retrieve this Secret yourself, as this task does it for you.

from prefect import Flow
from prefect.tasks.azure.blobstorage import BlobStorageDownload


azure_task = BlobStorageDownload(container="prefect", 
                                azure_credentials_secret="secret")

with Flow("example") as flow:
    res = azure_task()

Lastly, if you want to use a Secret when running a standalone task without the Flow context, you can use Secret from the client with .get():

from prefect.client.secrets import Secret

secret = Secret("azure").get()

some_task = SomeTaskThatNeedsSecretValueRatherThanSecretName(
    secret_value=secret,
)
some_task.run()
1 Like
3

Thanks for the detailed answer!