gmakstutis/cognee
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: Return async gather for documents (#1303)

Browse source 
<!-- .github/pull_request_template.md -->

## Description
Return async gather for documents

## DCO Affirmation
I affirm that all code in every commit of this pull request conforms to
the terms of the Topoteretes Developer Certificate of Origin.
This commit is contained in:
Igor Ilic 2025-08-27 17:08:41 +02:00 • committed by GitHub
commit 66673af56d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 17 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

58
cognee/modules/pipelines/operations/run_tasks.py
View file

@ -266,48 +266,24 @@ async def run_tasks(
if incremental_loading:
data = await resolve_data_directories(data)
# TODO: Return to using async.gather for data items after Cognee release
# # Create async tasks per data item that will run the pipeline for the data item
# data_item_tasks = [
# asyncio.create_task(
# _run_tasks_data_item(
# data_item,
# dataset,
# tasks,
# pipeline_name,
# pipeline_id,
# pipeline_run_id,
# context,
# user,
# incremental_loading,
# )
# )
# for data_item in data
# ]
# results = await asyncio.gather(*data_item_tasks)
# # Remove skipped data items from results
# results = [result for result in results if result]
### TEMP sync data item handling
results = []
# Run the pipeline for each data_item sequentially, one after the other
for data_item in data:
result = await _run_tasks_data_item(
data_item,
dataset,
tasks,
pipeline_name,
pipeline_id,
pipeline_run_id,
context,
user,
incremental_loading,
# Create async tasks per data item that will run the pipeline for the data item
data_item_tasks = [
asyncio.create_task(
_run_tasks_data_item(
data_item,
dataset,
tasks,
pipeline_name,
pipeline_id,
pipeline_run_id,
context,
user,
incremental_loading,
)
)
# Skip items that returned a false-y value
if result:
results.append(result)
### END
for data_item in data
]
results = await asyncio.gather(*data_item_tasks)
# Remove skipped data items from results
results = [result for result in results if result]