pipeline run logging

cognee/api/v1/add/add_v2.py

@@ -37,7 +37,7 @@ async def add(
 
     tasks = [Task(resolve_data_directories), Task(ingest_data, dataset_name, user)]
 
-    pipeline = run_tasks(tasks, data, "add_pipeline")
+    pipeline = run_tasks(tasks=tasks, dataset_id=None, data=data, pipeline_id="add_pipeline")
 
     async for result in pipeline:
         print(result)

cognee/api/v1/cognify/cognify_v2.py

@@ -73,8 +73,6 @@ async def cognify(
 async def run_cognify_pipeline(dataset: Dataset, user: User, tasks: list[Task]):
     data_documents: list[Data] = await get_dataset_data(dataset_id=dataset.id)
 
-    document_ids_str = [str(document.id) for document in data_documents]
-
     dataset_id = dataset.id
     dataset_name = generate_dataset_name(dataset.name)
 
@@ -90,15 +88,6 @@ async def run_cognify_pipeline(dataset: Dataset, user: User, tasks: list[Task]):
         logger.info("Dataset %s is already being processed.", dataset_name)
         return
 
-    await log_pipeline_status(
-        dataset_id,
-        PipelineRunStatus.DATASET_PROCESSING_STARTED,
-        {
-            "dataset_name": dataset_name,
-            "files": document_ids_str,
-        },
-    )
-
     try:
         if not isinstance(tasks, list):
             raise ValueError("Tasks must be a list")
@@ -117,27 +106,10 @@ async def run_cognify_pipeline(dataset: Dataset, user: User, tasks: list[Task]):
 
         send_telemetry("cognee.cognify EXECUTION COMPLETED", user.id)
 
-        await log_pipeline_status(
-            dataset_id,
-            PipelineRunStatus.DATASET_PROCESSING_COMPLETED,
-            {
-                "dataset_name": dataset_name,
-                "files": document_ids_str,
-            },
-        )
-
         return pipeline_run_status
+
     except Exception as error:
         send_telemetry("cognee.cognify EXECUTION ERRORED", user.id)
-
-        await log_pipeline_status(
-            dataset_id,
-            PipelineRunStatus.DATASET_PROCESSING_ERRORED,
-            {
-                "dataset_name": dataset_name,
-                "files": document_ids_str,
-            },
-        )
         raise error
 
 

cognee/modules/pipelines/models/PipelineRun.py

@@ -1,7 +1,7 @@
 import enum
 from uuid import uuid4
 from datetime import datetime, timezone
-from sqlalchemy import Column, DateTime, JSON, Enum, UUID
+from sqlalchemy import Column, DateTime, JSON, Enum, UUID, String
 from cognee.infrastructure.databases.relational import Base
 
 
@@ -20,5 +20,5 @@ class PipelineRun(Base):
 
     status = Column(Enum(PipelineRunStatus))
 
-    pipeline_id = Column(UUID, index=True)
+    pipeline_id = Column(String, index=True)
     run_info = Column(JSON)

cognee/modules/pipelines/operations/get_pipeline_status.py

@@ -14,12 +14,12 @@ async def get_pipeline_status(pipeline_ids: list[UUID]):
                 PipelineRun,
                 func.row_number()
                 .over(
-                    partition_by=PipelineRun.run_id,
+                    partition_by=PipelineRun.id,
                     order_by=PipelineRun.created_at.desc(),
                 )
                 .label("rn"),
             )
-            .filter(PipelineRun.run_id.in_(pipeline_ids))
+            .filter(PipelineRun.id.in_(pipeline_ids))
             .subquery()
         )
 
@@ -29,7 +29,7 @@ async def get_pipeline_status(pipeline_ids: list[UUID]):
 
         runs = (await session.execute(latest_runs)).scalars().all()
 
-        pipeline_statuses = {str(run.run_id): run.status for run in runs}
+        pipeline_statuses = {str(run.id): run.status for run in runs}
 
         return pipeline_statuses
 

cognee/modules/pipelines/operations/logPipelineRunComplete.py

@@ -12,8 +12,8 @@ async def logPipelineRunComplete(pipeline_id: UUID, dataset_id: UUID, data: list
         pipeline_id=pipeline_id,
         status=PipelineRunStatus.DATASET_PROCESSING_COMPLETED,
         run_info={
-            "dataset_id": dataset_id,
-            "data": [data.id for data in data],
+            "dataset_id": str(dataset_id),
+            "data": [str(data.id) for data in data] if isinstance(data, list) else data,
         },
     )
 

cognee/modules/pipelines/operations/logPipelineRunError.py

@@ -4,16 +4,17 @@
 from cognee.modules.pipelines.models import PipelineRun, PipelineRunStatus
 
 
-async def logPipelineRunError(pipeline_id: UUID, dataset_id: UUID, data: list[Data]):
+async def logPipelineRunError(pipeline_id: str, dataset_id: UUID, data: list[Data], e: Exception):
     pipeline_run_id = uuid4()
 
     pipeline_run = PipelineRun(
         id=pipeline_run_id,
         pipeline_id=pipeline_id,
         status=PipelineRunStatus.DATASET_PROCESSING_ERRORED,
         run_info={
-            "dataset_id": dataset_id,
-            "data": [data.id for data in data],
+            "dataset_id": str(dataset_id),
+            "data": [str(data.id) for data in data] if isinstance(data, list) else data,
+            "error": str(e),
         },
     )
 

cognee/modules/pipelines/operations/logPipelineRunStart.py

@@ -4,16 +4,16 @@
 from cognee.modules.pipelines.models import PipelineRun, PipelineRunStatus
 
 
-async def logPipelineRunStart(pipeline_id: UUID, dataset_id: UUID, data: list[Data]):
+async def logPipelineRunStart(pipeline_id: str, dataset_id: UUID, data: list[Data]):
     pipeline_run_id = uuid4()
 
     pipeline_run = PipelineRun(
         id=pipeline_run_id,
         pipeline_id=pipeline_id,
         status=PipelineRunStatus.DATASET_PROCESSING_STARTED,
         run_info={
-            "dataset_id": dataset_id,
-            "data": [data.id for data in data],
+            "dataset_id": str(dataset_id),
+            "data": [str(data.id) for data in data] if isinstance(data, list) else data,
         },
     )
 

cognee/modules/pipelines/operations/run_tasks.py

@@ -276,7 +276,7 @@ async def run_tasks(tasks: list[Task], dataset_id: UUID, data: list[Data], pipel
         async for _ in run_tasks_with_telemetry(tasks, data, pipeline_id):
             pass
 
-        yield await logPipelineRunComplete(pipeline_run.id, data)
+        yield await logPipelineRunComplete(pipeline_id, dataset_id, data)
     except Exception as e:
-        yield await logPipelineRunError(pipeline_run.id, tasks, data, e)
+        yield await logPipelineRunError(pipeline_id, dataset_id, data, e)
         raise e