Enhance feedback collection flow with Prefect Cloud integration and improved error handling

app.py

@@ -1,8 +1,11 @@
 import asyncio
+import os
 import gradio as gr
 from prefect import flow, get_run_logger, pause_flow_run, settings
 from prefect.context import get_run_context
 from prefect.input import RunInput
+from prefect.deployments import Deployment
+from prefect.server.schemas.schedules import IntervalSchedule
 
 
 class UserFeedback(RunInput):
@@ -15,22 +18,47 @@ class UserFeedback(RunInput):
 
 @flow(name="interactive-feedback-flow")
 async def collect_feedback(username: str):
-    """A flow that pauses to collect user feedback via the Prefect UI."""
+    """A flow that pauses to collect user feedback via the Prefect Cloud UI."""
     logger = get_run_logger()
 
+    # Ensure we're using Prefect Cloud
+    logger.info("Running with Prefect Cloud configuration")
+
     logger.info(f"Started feedback collection for user: {username}")
 
     # Construct a message that would typically be sent as a notification
-    message = f"Collecting feedback from user: {username}"
-
-    # Get flow run context for UI URL
+    message = (
+        f"Collecting feedback from user: {username}"  # Get flow run context for UI URL
+    )
     flow_run = get_run_context().flow_run
     flow_run_url = ""
 
-    if flow_run and settings.PREFECT_UI_URL:
-        flow_run_url = (
-            f"{settings.PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run.id}"
-        )
+    if flow_run:
+        # For Prefect Cloud, use the PREFECT_API_URL to determine the Cloud UI URL
+        if os.environ.get("PREFECT_API_URL") and "api.prefect.cloud" in os.environ.get(
+            "PREFECT_API_URL", ""
+        ):
+            # Extract account_id and workspace_name from the API URL
+            # Format: https://api.prefect.cloud/api/accounts/{account_id}/workspaces/{workspace_name}/
+            api_url = os.environ["PREFECT_API_URL"]
+            parts = api_url.split("/")
+
+            if (
+                len(parts) >= 7
+            ):  # Ensure we have enough parts to extract account_id and workspace_name
+                account_id = parts[5]
+                workspace_name = parts[7]
+                cloud_ui_url = f"https://app.prefect.cloud/account/{account_id}/workspace/{workspace_name}"
+                flow_run_url = f"{cloud_ui_url}/flow-runs/flow-run/{flow_run.id}"
+            else:
+                # Fallback to standard URL formation if we can't parse the API URL
+                if settings.PREFECT_UI_URL:
+                    flow_run_url = f"{settings.PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run.id}"
+        else:
+            # Fallback for local Prefect or other setups
+            if settings.PREFECT_UI_URL:
+                flow_run_url = f"{settings.PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run.id}"
+
         logger.info(f"Flow run UI available at: {flow_run_url}")
 
     # Pause the flow and wait for input
@@ -42,7 +70,7 @@ async def collect_feedback(username: str):
             comment="",
             improve_feature=False,
         ),
-        timeout=300,  # 5 minutes timeout
+        timeout=60,  # 5 minutes timeout
     )
 
     # Process the input received after the flow is resumed
@@ -66,21 +94,80 @@ async def collect_feedback(username: str):
 def run_feedback_flow(username):
     """Run the Prefect flow and handle the UI interaction."""
     try:
-        # Run the flow and get the result
-        result = asyncio.run(collect_feedback(username))
+        if not username:
+            return "Please enter a username before starting the process."
+
+        print(f"Starting feedback collection for user: {username}")
+
+        # Create a deployment of the flow to ensure it runs properly in Prefect Cloud
+        from prefect.deployments import Deployment
+
+        # Create or update the deployment
+        deployment = Deployment.build_from_flow(
+            flow=collect_feedback,
+            name="interactive-feedback-deployment",
+            version=None,  # Auto-version
+            work_queue_name="default",
+            tags=["interactive", "feedback"],
+            parameters={"username": username},
+            enforce_parameter_schema=False,
+        )
+
+        # Apply the deployment
+        deployment_id = deployment.apply()
+        print(f"Created deployment with ID: {deployment_id}")
+
+        # Run the deployment
+        from prefect.deployments import run_deployment
 
-        # Format the result as a string for display
-        message = f"""
-        Thank you, {result['username']}!
+        flow_run = run_deployment(
+            name="interactive-feedback-flow/interactive-feedback-deployment",
+            parameters={"username": username},
+        )
+
+        print(f"Flow run created with ID: {flow_run.id}")
+
+        # Get the URL
+        flow_run_url = ""
+        if settings.PREFECT_UI_URL:
+            flow_run_url = (
+                f"{settings.PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run.id}"
+            )
+        elif os.environ.get(
+            "PREFECT_API_URL"
+        ) and "api.prefect.cloud" in os.environ.get("PREFECT_API_URL", ""):
+            # Extract account_id and workspace_name from the API URL
+            api_url = os.environ["PREFECT_API_URL"]
+            parts = api_url.split("/")
+
+            if len(parts) >= 7:
+                account_id = parts[5]
+                workspace_name = parts[7]
+                flow_run_url = f"https://app.prefect.cloud/account/{account_id}/workspace/{workspace_name}/flow-runs/flow-run/{flow_run.id}"
+
+        print(f"Flow run UI available at: {flow_run_url}")
+
+        return f"""
+        Process started for {username}!
         
-        Your feedback has been recorded:
-        - Rating: {result['rating']}/5
-        - Comment: {result['comment']}
-        - Feature improvement requested: {result['improve_feature']}
+        IMPORTANT INSTRUCTIONS:
+        
+        Go to this URL to provide your feedback:
+        {flow_run_url}
+        
+        1. Click the "Resume" button in the Prefect UI
+        2. Fill out the feedback form and submit
+        3. Your input will be processed by the workflow
+        
+        Note: This window will not automatically update with the results.
+        The complete results will be available in the Prefect UI and terminal.
         """
 
-        return message
     except Exception as e:
+        import traceback
+
+        print(f"Error: {str(e)}")
+        print(traceback.format_exc())
         return f"An error occurred: {str(e)}"