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

move config getter into cognee/modules/pipelines/operations/run_tasks.py and make the indentation a bit more readable

Browse source
This commit is contained in:
Rita Aleksziev 2024-12-06 13:38:54 +01:00
parent dbfa91b635
commit 462fcef240
1 changed files with 19 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -161,25 +161,29 @@ async def run_tasks_base(tasks: list[Task], data = None, user: User = None):
})
raise error
async def run_tasks_with_telemetry(tasks: list[Task], data, pipeline_name: str, config: dict):
async def run_tasks_with_telemetry(tasks: list[Task], data, pipeline_name: str):
config = get_current_settings()
logger.debug("\nRunning pipeline with configuration:\n%s\n", json.dumps(config, indent = 1))
user = await get_default_user()
try:
logger.info("Pipeline run started: `%s`", pipeline_name)
send_telemetry("Pipeline Run Started", user.id, {
"pipeline_name": pipeline_name,
} | config
)
send_telemetry("Pipeline Run Started",
user.id,
additional_properties = {"pipeline_name": pipeline_name, } | config
)
async for result in run_tasks_base(tasks, data, user):
yield result
logger.info("Pipeline run completed: `%s`", pipeline_name)
send_telemetry("Pipeline Run Completed", user.id, {
"pipeline_name": pipeline_name,
})
send_telemetry("Pipeline Run Completed",
user.id,
additional_properties = {"pipeline_name": pipeline_name, }
)
except Exception as error:
logger.error(
"Pipeline run errored: `%s`\n%s\n",
@ -187,14 +191,14 @@ async def run_tasks_with_telemetry(tasks: list[Task], data, pipeline_name: str,
str(error),
exc_info = True,
)
send_telemetry("Pipeline Run Errored", user.id, {
"pipeline_name": pipeline_name,
})
send_telemetry("Pipeline Run Errored",
user.id,
additional_properties = {"pipeline_name": pipeline_name, } | config
)
raise error
async def run_tasks(tasks: list[Task], data = None, pipeline_name: str = "default_pipeline"):
config = get_current_settings()
async for result in run_tasks_with_telemetry(tasks, data, pipeline_name, config):
async for result in run_tasks_with_telemetry(tasks, data, pipeline_name):
yield result