busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

semantic checker script update

Browse Source
This commit is contained in:
busya
2026-01-13 17:33:57 +03:00
parent b2529973eb
commit 11c59fb420
11 changed files with 3013 additions and 1044 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
backend/src/core/task_manager/persistence.py
View File

@@ -21,11 +21,16 @@ from ..logger import logger, belief_scope
# @SEMANTICS: persistence, service, database, sqlalchemy
# @PURPOSE: Provides methods to save and load tasks from the tasks.db database using SQLAlchemy.
class TaskPersistenceService:
# [DEF:__init__:Function]
# @PURPOSE: Initializes the persistence service.
# @PRE: None.
# @POST: Service is ready.
def __init__(self):
# We use TasksSessionLocal from database.py
pass
# [/DEF:__init__:Function]
# [DEF:TaskPersistenceService.persist_task:Function]
# [DEF:persist_task:Function]
# @PURPOSE: Persists or updates a single task in the database.
# @PARAM: task (Task) - The task object to persist.
def persist_task(self, task: Task) -> None:
@@ -66,17 +71,17 @@ class TaskPersistenceService:
logger.error(f"Failed to persist task {task.id}: {e}")
finally:
session.close()
# [/DEF:TaskPersistenceService.persist_task:Function]
# [/DEF:persist_task:Function]
# [DEF:TaskPersistenceService.persist_tasks:Function]
# [DEF:persist_tasks:Function]
# @PURPOSE: Persists multiple tasks.
# @PARAM: tasks (List[Task]) - The list of tasks to persist.
def persist_tasks(self, tasks: List[Task]) -> None:
for task in tasks:
self.persist_task(task)
# [/DEF:TaskPersistenceService.persist_tasks:Function]
# [/DEF:persist_tasks:Function]
# [DEF:TaskPersistenceService.load_tasks:Function]
# [DEF:load_tasks:Function]
# @PURPOSE: Loads tasks from the database.
# @PARAM: limit (int) - Max tasks to load.
# @PARAM: status (Optional[TaskStatus]) - Filter by status.
@@ -119,9 +124,9 @@ class TaskPersistenceService:
return loaded_tasks
finally:
session.close()
# [/DEF:TaskPersistenceService.load_tasks:Function]
# [/DEF:load_tasks:Function]
# [DEF:TaskPersistenceService.delete_tasks:Function]
# [DEF:delete_tasks:Function]
# @PURPOSE: Deletes specific tasks from the database.
# @PARAM: task_ids (List[str]) - List of task IDs to delete.
def delete_tasks(self, task_ids: List[str]) -> None:
@@ -137,7 +142,7 @@ class TaskPersistenceService:
logger.error(f"Failed to delete tasks: {e}")
finally:
session.close()
# [/DEF:TaskPersistenceService.delete_tasks:Function]
# [/DEF:delete_tasks:Function]
# [/DEF:TaskPersistenceService:Class]
# [/DEF:TaskPersistenceModule:Module]