Adding the ID of the producer pipeline run to artifact versions

src/zenml/models/v2/core/artifact_version.py

@@ -137,6 +137,9 @@ class ArtifactVersionResponseBody(WorkspaceScopedResponseBody):
     tags: List[TagResponse] = Field(
         title="Tags associated with the model",
     )
+    producer_pipeline_run_id: Optional[UUID] = Field(
+        title="The ID of the pipeline run that generated this artifact version."
+    )
 
     _convert_source = convert_source_validator("materializer", "data_type")
 
@@ -223,6 +226,15 @@ def tags(self) -> List[TagResponse]:
         """
         return self.get_body().tags
 
+    @property
+    def producer_pipeline_run_id(self) -> Optional[UUID]:
+        """The `producer_pipeline_run_id` property.
+
+        Returns:
+            the value of the property.
+        """
+        return self.get_body().producer_pipeline_run_id
+
     @property
     def artifact_store_id(self) -> Optional[UUID]:
         """The `artifact_store_id` property.

src/zenml/zen_stores/schemas/artifact_schemas.py

@@ -297,6 +297,15 @@ def to_model(self, hydrate: bool = False) -> ArtifactVersionResponse:
             # This is an old source which was an importable source path
             data_type = Source.from_import_path(self.data_type)
 
+        producer_step_run_id, producer_pipeline_run_id = None, None
+        if self.output_of_step_runs:
+            step_run = self.output_of_step_runs[0].step_run
+            if step_run.status == ExecutionStatus.COMPLETED:
+                producer_step_run_id = step_run.id
+                producer_pipeline_run_id = step_run.pipeline_run_id
+            else:
+                producer_step_run_id = step_run.original_step_run_id
+
         # Create the body of the model
         body = ArtifactVersionResponseBody(
             artifact=self.artifact.to_model(),
@@ -309,19 +318,12 @@ def to_model(self, hydrate: bool = False) -> ArtifactVersionResponse:
             created=self.created,
             updated=self.updated,
             tags=[t.tag.to_model() for t in self.tags],
+            producer_pipeline_run_id=producer_pipeline_run_id,
         )
 
         # Create the metadata of the model
         metadata = None
         if hydrate:
-            producer_step_run_id = None
-            if self.output_of_step_runs:
-                step_run = self.output_of_step_runs[0].step_run
-                if step_run.status == ExecutionStatus.COMPLETED:
-                    producer_step_run_id = step_run.id
-                else:
-                    producer_step_run_id = step_run.original_step_run_id
-
             metadata = ArtifactVersionResponseMetadata(
                 workspace=self.workspace.to_model(),
                 artifact_store_id=self.artifact_store_id,