brian
02/22/2022, 5:21 PMbrian
02/22/2022, 5:21 PMMU_SENTRY_DSNKevin 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_stateKevin 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 prefectbrian
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