@dramatiq.actor(time_limit=2 * 60 * 60 * 1000, abortable=True, max_retries=0)
async def todo_task(
task_name
):
session_id = str(uuid.uuid4())
context = set_session_context(session_id=session_id)
try:
# Attempt to find and abort any previously running tasks with the same task_name
already_running_tasks = await TaskService().get_already_running_task_by_task_name(
task_name=task_name
)
logger.info(f"Found tasks to abort: {already_running_tasks}")
if already_running_tasks:
for id in already_running_tasks:
logger.info(f"Attempting to abort task with ID: {id}")
abort(message_id=id)
logger.info(f"Sent abort signal to task with ID: {id}")
logger.info(f"Started partial indexing with {task_id}, session_id:{session_id}")
# Run the main task logic
await sometask().todo()
logger.info(f"Completed partial indexing with {task_id}, session_id:{session_id}")
except Abort:
logger.warning(f"Task aborted")
except Exception as e:
logger.error(
f"Error"
)
finally:
await session.close()
reset_session_context(context=context)
