Martin T
03/09/2022, 12:22 PMPrefectSecret()
to load secrets from Prefect Cloud. Is it possible to load secret from cloud into prefect.context.secrets
during flow start so all tasks can access them when needed?
Most of our tasks variables are auth-related, and the code/layout is getting overly complex.Anna Geller
03/09/2022, 12:55 PMfrom prefect import task, Flow
from prefect.client.secrets import Secret
@task(log_stdout=True)
def use_secret_within_some_task_directly():
secret_value = Secret("A_TEST_SECRET").get()
print(secret_value)
with Flow("secret_flow_client") as flow:
use_secret_within_some_task_directly()
if __name__ == "__main__":
flow.run()
But best practice is to pass it to downstream tasks as data dependency as shown here:
from prefect import task, Flow
from prefect.tasks.secrets import PrefectSecret
@task(log_stdout=True)
def use_secret_from_data_dependency(secret_value: str):
print(secret_value)
with Flow("secret_flow_client") as flow:
secret_val = PrefectSecret("A_TEST_SECRET")
use_secret_from_data_dependency(secret_val)
if __name__ == "__main__":
flow.run()
In both cases, you don't need to store secrets in the context - storing secrets in the context is only needed:
1. If you are using Prefect Server since Server doesn't have the Secret backend
2. For local execution - even though local flow runs may retrieve Secrets from the Cloud backend, too.Martin T
03/09/2022, 1:13 PMSecret("A_TEST_SECRET").get()
was what I was looking for. I wasn't sure if this was retrieving from Prefect Cloud Secrets. https://docs.prefect.io/orchestration/concepts/secrets.html#using-secrets-elsewhere
Why is Secret().get()
discouraged?Anna Geller
03/09/2022, 1:38 PMfrom prefect import task, Flow
from prefect.client.secrets import Secret
secret_value = Secret("A_TEST_SECRET").get()
Then, this is problematic since secrets are generally considered a runtime information, while setting this as above ๐ retrieves the secret at build time during flow registration. As long as you use script storage, even this is fine, but if you use the default pickle storage and you retrieve the Secret globally like this, then your pickled flow would become a security risk as it would be possible to get the value of your Secret from that pickle file.
So doing this ๐ within a task is not an anti-pattern and is still fine because you retrieve this at runtime within the task run:
@task(log_stdout=True)
def use_secret_within_some_task_directly():
secret_value = Secret("A_TEST_SECRET").get()
print(secret_value)
So TL;DR - donโt use prefect.client.secrets.Secret
globally when using pickle storage - in all other use cases, it should be fine.Kevin Kho
03/09/2022, 2:34 PMflow.storage = StorageClass(..., secrets=["SECRET_NAME"])