Adds more detail to task run state change events, including the full …

…task run (#14626)
PrefectHQ · Jul 16, 2024 · 387a2a4 · 387a2a4
1 parent 567394f
commit 387a2a4
Show file tree

Hide file tree

Showing 3 changed files with 369 additions and 67 deletions.
diff --git a/src/prefect/client/schemas/objects.py b/src/prefect/client/schemas/objects.py
@@ -791,7 +791,7 @@ class TaskRun(ObjectBaseModel):
 
     state: Optional[State] = Field(
         default=None,
-        description="The state of the flow run.",
+        description="The state of the task run.",
         examples=["State(type=StateType.COMPLETED)"],
     )
 

diff --git a/src/prefect/utilities/engine.py b/src/prefect/utilities/engine.py
@@ -744,6 +744,12 @@ def emit_task_run_state_change_event(
                     "message": truncated_to(
                         state_message_truncation_length, initial_state.message
                     ),
+                    "state_details": initial_state.state_details.model_dump(
+                        mode="json",
+                        exclude_none=True,
+                        exclude_unset=True,
+                        exclude={"flow_run_id", "task_run_id"},
+                    ),
                 }
                 if initial_state
                 else None
@@ -754,7 +760,30 @@ def emit_task_run_state_change_event(
                 "message": truncated_to(
                     state_message_truncation_length, validated_state.message
                 ),
+                "state_details": validated_state.state_details.model_dump(
+                    mode="json",
+                    exclude_none=True,
+                    exclude_unset=True,
+                    exclude={"flow_run_id", "task_run_id"},
+                ),
+                "data": validated_state.data.model_dump(mode="json")
+                if isinstance(validated_state.data, BaseResult)
+                else None,
             },
+            "task_run": task_run.model_dump(
+                mode="json",
+                exclude_none=True,
+                exclude={
+                    "id",
+                    "created",
+                    "updated",
+                    "flow_run_id",
+                    "state_id",
+                    "state_type",
+                    "state_name",
+                    "state",
+                },
+            ),
         },
         resource={
             "prefect.resource.id": f"prefect.task-run.{task_run.id}",