Amogh Kulkarni
04/22/2022, 4:20 PMException raised while calling state handlers: ClientError([{'path': ['secret_value'], 'message': 'An unknown error occurred.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}])
Traceback (most recent call last):
File "/usr/local/lib/python3.9/site-packages/prefect/client/secrets.py", line 137, in get
value = secrets[self.name]
KeyError: 'SLACK_WEBHOOK_URL'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/usr/local/lib/python3.9/site-packages/prefect/engine/cloud/task_runner.py", line 64, in call_runner_target_handlers
new_state = super().call_runner_target_handlers(
File "/usr/local/lib/python3.9/site-packages/prefect/engine/task_runner.py", line 113, in call_runner_target_handlers
new_state = handler(self.task, old_state, new_state) or new_state
File "/usr/local/lib/python3.9/site-packages/toolz/functoolz.py", line 306, in __call__
return self._partial(*args, **kwargs)
File "/usr/local/lib/python3.9/site-packages/prefect/utilities/notifications/notifications.py", line 299, in slack_notifier
str, prefect.client.Secret(webhook_secret or "SLACK_WEBHOOK_URL").get()
File "/usr/local/lib/python3.9/site-packages/prefect/client/secrets.py", line 161, in get
raise exc
File "/usr/local/lib/python3.9/site-packages/prefect/client/secrets.py", line 145, in get
result = self.client.graphql(
File "
/usr/local/lib/python3.9/site-packages/prefect/client/client.py", line 570, in graphql
raise ClientError(result["errors"])
prefect.exceptions.ClientError: [{'path': ['secret_value'], 'message': 'An unknown error occurred.', 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]
03:02:29
IKevin Kho
with Flow(...) as flow:
PrefectSecret("SLACK_WEBHOOK_URL".max_retries=3)
Amogh Kulkarni
04/22/2022, 4:41 PM@task(log_stdout=True, state_handlers=[handler],name=f"truncate_snowflake_play_table@{dag_name}",)
def truncate_snowflake_play_table():
sfq = <function to connecto to snowflake>
sfq.run(query=<sql query to execute in snowflake>")
Kevin Kho
flow.storage = ["SLACK_WEBHOOK_SECRET"]
that way, this secret is only pulled once. If you map over a lot of these tasks with the state handler, it will be pulled each time and some of those API calls may failAmogh Kulkarni
04/27/2022, 12:59 AMKevin Kho
def mystatehandler(..):
Secret("SLACK_WEBHOOK_URL").get()
that will pull for each task or each failed task.Amogh Kulkarni
04/27/2022, 1:14 AMKevin Kho
flow.storage = SomeStorage(..., secrets=["SLACK_WEBHOOK_SECRET"])
just add thisAmogh Kulkarni
04/29/2022, 5:21 PMRainer Schülke
05/10/2022, 8:13 AMKevin Kho