brian
02/22/2022, 5:21 PMbrian
02/22/2022, 5:21 PMMU_SENTRY_DSN
Kevin Kho
Secret(…).get()
will inside a state handler. Or maybe PrefectSecret().run()
brian
02/22/2022, 5:22 PMimport prefect
prefect.context.setdefault("secrets", {}) # to make sure context has a secrets attribute
prefect.context.secrets["MY_KEY"] = "MY_VALUE"
brian
02/22/2022, 5:22 PMbrian
02/22/2022, 5:22 PMdef _handle_flow_states(obj, old_state, new_state):
if new_state.is_running():
logger = prefect.context.get("logger")
<http://logger.info|logger.info>("flow is running, initializing sentry sdk")
# make sure context has a secrets attribute
# see this <https://docs.prefect.io/api/latest/client/secrets.html#secret>
prefect.context.setdefault("secrets", {})
sentry_sdk.init(dsn=prefect.context.secrets["MU_SENTRY_DSN"])
<http://logger.info|logger.info>("initialized sentry sdk")
return new_state
Kevin Kho
brian
02/22/2022, 5:24 PMsetdefault
call do nothing? i.e. prefect.context.secrets should already be a dict?Kevin Kho
import os
os.environ["PREFECT__CONTEXT__SECRETS__MY_KEY"]
import prefect
brian
02/22/2022, 5:25 PMKevin Kho
flow.storage = SomeStorage(...,secrets=[])
brian
02/22/2022, 5:25 PMbrian
02/22/2022, 5:26 PMbrian
02/22/2022, 5:26 PMbrian
02/22/2022, 5:29 PMflow.storage = GitHub(
repo="...",
path="...",
secrets=["MU_SENTRY_DSN"],
)
Kevin Kho
PrefectSecret
like
with Flow() as flow:
PrefectSecret("something")
this is pulled during runtime because it’s a task and the task has deferred execution
So if you do it in a state handler, to pull it immediately:
def mystatehandler(task, old_state, new_state):
PrefectSecret("something").run()
# or
Secret("something").get()
I am less sure PrefectSecret will work in the state handler. Secret.get() is preferred outside of the FlowKevin Kho