Repository: topoteretes/cognee
File: cognee/modules/pipelines/operations/run_tasks.py (Python)

Code before:
        yield {
            "run_info": PipelineRunErrored(
                pipeline_run_id=pipeline_run_id,
                payload=error,
                dataset_id=dataset.id,
                dataset_name=dataset.name,
            )
        }


async def run_tasks_per_data(
    data_item,
    dataset,
    tasks,
    pipeline_name,
    pipeline_id,
    pipeline_run_id,
    context,
    user,
    incremental_loading,
):
    # Go through async generator and return data item processing result. Result can be None when data item is skipped,
    # PipelineRunCompleted when processing was successful and PipelineRunErrored if there were issues
    result = None
    if incremental_loading:
        async for result in run_tasks_incremental_data_item_generator(
            data_item=data_item,
            dataset=dataset,
            tasks=tasks,
            pipeline_name=pipeline_name,
            pipeline_id=pipeline_id,
            pipeline_run_id=pipeline_run_id,
            context=context,
            user=user,
        ):
            pass
    else:
        async for result in run_tasks_data_item_generator(
            data_item=data_item,
            dataset=dataset,
            tasks=tasks,
            pipeline_id=pipeline_id,
            pipeline_run_id=pipeline_run_id,
            context=context,
            user=user,
        ):
            pass

    return result



Review comment (suggestion):
It is a bit confusing name for the function, maybe `run_tasks_incremental_data_item_generator ` -> `run_tasks_incremental`. Would be good that name tells us what it does.

Code after:
            yield PipelineRunYield(
                pipeline_run_id=pipeline_run_id,
                dataset_id=dataset.id,
                dataset_name=dataset.name,
                payload=result,
            )

        yield {
            "run_info": PipelineRunCompleted(
                pipeline_run_id=pipeline_run_id,
                dataset_id=dataset.id,
                dataset_name=dataset.name,
            )
        }

    async def _run_tasks_data_item(
        data_item,
        dataset,
        tasks,
        pipeline_name,
        pipeline_id,
        pipeline_run_id,
        context,
        user,
        incremental_loading,
    ):
        # Go through async generator and return data item processing result. Result can be PipelineRunAlreadyCompleted when data item is skipped,
        # PipelineRunCompleted when processing was successful and PipelineRunErrored if there were issues
        result = None
        if incremental_loading:
            async for result in _run_tasks_data_item_incremental(
                data_item=data_item,
                dataset=dataset,
                tasks=tasks,
                pipeline_name=pipeline_name,
                pipeline_id=pipeline_id,
                pipeline_run_id=pipeline_run_id,
                context=context,
                user=user,
            ):
                pass
        else:
            async for result in _run_tasks_data_item_regular(
                data_item=data_item,
                dataset=dataset,
                tasks=tasks,
                pipeline_id=pipeline_id,
                pipeline_run_id=pipeline_run_id,
                context=context,
                user=user,
            ):