from argo.workflows.client import V1alpha1Api
from argo.workflows.client.rest import ApiException
def submit_argo_workflow():
api_instance = V1alpha1Api()
try:
api_response = api_instance.create_namespaced_workflow(...)
# 标记通知已发送
...
except ApiException as e:
print("Exception: %s\n" % e)
# 标记通知发送失败
...
请注意,在上面的代码中,我们使用try-except块来捕获可能出现的异常并进行处理。在api_response成功时,标志所需的通知已成功发送,反之,我们将通知标为发送失败。
def send_notification(api_instance, notification_data):
try:
api_instance.create_namespaced_notification(...)
return True
except ApiException as e:
print("Exception when calling V1alpha1Api->create_namespaced_notification: %s\n" % e)
return False
注意,在上面的代码中,我们使用一个名为notification_data的参数来传递通知所需的数据。这个数据可以被传递给create_namespaced_notification或其他Argo API。
请参考Argo文档中的有关更多信息:https://argoproj.github.io/docs/argo-events/notifications/
上一篇:Argo通知已发送的条件问题