Robert Denham
06/30/2025, 5:46 AMMarvin
06/30/2025, 5:46 AMMarvin
06/30/2025, 5:48 AMpython
from prefect import flow, task
from prefect.utilities.urls import url_for
@task
def get_task_dashboard_url(task_run_id: str) -> str:
    """Get the UI dashboard URL for a specific task run."""
    return url_for(type="task-run", id=task_run_id)
@flow
def create_failure_summary(failed_task_runs: list[str]) -> str:
    """Create a summary of failed tasks with their dashboard URLs."""
    summary_parts = []
    
    for task_run_id in failed_task_runs:
        dashboard_url = get_task_dashboard_url(task_run_id)
        summary_parts.append(f"Failed Task Run: {task_run_id}")
        summary_parts.append(f"Dashboard URL: {dashboard_url}")
        summary_parts.append("---")
    
    return "\n".join(summary_parts)
# Example usage
@flow
def main_flow():
    failed_runs = ["task-run-id-1", "task-run-id-2"]  # Replace with actual task run IDs
    failure_summary = create_failure_summary(failed_runs)
    # Use the summary in your email
    print(failure_summary)url_for()prefect.utilities.urlspython
from prefect.runtime import task_run
@task
def some_task():
    # Get the current task's dashboard URL
    current_task_url = task_run.ui_url
    return current_task_url<https://your-prefect-server/task-runs/task-run-id>