ss-tools/backend/src/scripts/delete_running_tasks.py

#!/usr/bin/env python3
# #region DeleteRunningTasksUtil [C:3] [TYPE Module]
# @defgroup Module Module group.
# @PURPOSE Script to delete tasks with RUNNING status from the database.
# @LAYER Infrastructure
# @SEMANTICS maintenance, database, cleanup
# @RELATION DEPENDS_ON ->[TaskRecord]

from sqlalchemy.orm import Session

from src.core.database import TasksSessionLocal
from src.models.task import TaskRecord

# #region delete_running_tasks [C:4] [TYPE Function]
# @ingroup Module
# @PURPOSE Delete all tasks with RUNNING status from the database.
# @PRE Database is accessible and TaskRecord model is defined.
# @POST All tasks with status 'RUNNING' are removed from the database.
# @SIDE_EFFECT Modifies DB: deletes RUNNING tasks. Prints to stdout.

def delete_running_tasks():
    """Delete all tasks with RUNNING status from the database."""
    session: Session = TasksSessionLocal()
    try:
        # Find all task records with RUNNING status
        running_tasks = session.query(TaskRecord).filter(TaskRecord.status == "RUNNING").all()

        if not running_tasks:
            print("No RUNNING tasks found.")
            return

        print(f"Found {len(running_tasks)} RUNNING tasks:")
        for task in running_tasks:
            print(f"- Task ID: {task.id}, Type: {task.type}")

        # Delete the found tasks
        session.query(TaskRecord).filter(TaskRecord.status == "RUNNING").delete(synchronize_session=False)
        session.commit()

        print(f"Successfully deleted {len(running_tasks)} RUNNING tasks.")
    except Exception as e:
        session.rollback()
        print(f"Error deleting tasks: {e}")
    finally:
        session.close()
# #endregion delete_running_tasks

if __name__ == "__main__":
    delete_running_tasks()
# #endregion DeleteRunningTasksUtil