Sergey Moskvin
01/28/2025, 7:40 PMMarvin
01/28/2025, 7:41 PMprefect.runtime.task_run.TaskRunContext
.
2. Construct the URL: Use the task run ID to construct the URL to the Prefect Server UI. The URL format typically follows a pattern like http://<your-prefect-server>/task-run/<task-run-id>
.
3. Send Notification: Use your preferred method to send a notification (e.g., email, Slack) with the constructed URL when the task fails.
Here is a basic example:
python
from prefect import task, get_run_logger
from prefect.runtime.task_run import TaskRunContext
@task
def my_task():
context = TaskRunContext.get()
task_run_id = context.task_run_id
prefect_server_url = "http://<your-prefect-server>/task-run/"
task_run_url = f"{prefect_server_url}{task_run_id}"
try:
# Your task logic here
pass
except Exception as e:
logger = get_run_logger()
logger.error(f"Task failed. View logs at {task_run_url}")
# Send notification with task_run_url
raise e
Replace <your-prefect-server>
with your actual Prefect Server address.
For more detailed information, you can refer to the Prefect documentation and the task engine source code.Nate
01/28/2025, 8:15 PMruntime.flow_run
but not for tasks, so I've opened this PR which will allow for the following
from prefect import task
import prefect.runtime
import webbrowser
@task
def open_ui():
webbrowser.open_new_tab(prefect.runtime.task_run.ui_url)
open_ui()