35 lines
1.0 KiB
Python
35 lines
1.0 KiB
Python
from database.tscheduler.model import TScheduler
|
|
|
|
def create_task(db, task: TScheduler):
|
|
db.add(task)
|
|
db.commit()
|
|
db.refresh(task)
|
|
return task
|
|
|
|
def get_task_by_id(db, task_id: int):
|
|
return db.query(TScheduler).filter(TScheduler.id == task_id).first()
|
|
|
|
def get_active_tasks(db):
|
|
return db.query(TScheduler).filter(TScheduler.active == True).all()
|
|
|
|
def get_tasks_by_executor(db, executor: str):
|
|
return db.query(TScheduler).filter(
|
|
TScheduler.executor == executor,
|
|
TScheduler.active == True
|
|
).all()
|
|
|
|
def update_task(db, task_id: int, updates: dict):
|
|
task = db.query(TScheduler).filter(TScheduler.id == task_id).first()
|
|
if task:
|
|
for key, value in updates.items():
|
|
setattr(task, key, value)
|
|
db.commit()
|
|
db.refresh(task)
|
|
return task
|
|
|
|
def delete_task(db, task_id: int):
|
|
task = db.query(TScheduler).filter(TScheduler.id == task_id).first()
|
|
if task:
|
|
db.delete(task)
|
|
db.commit()
|
|
return task |