Fix update and delete event

Author: wu-clan

backend/app/task/crud/crud_scheduler.py

@@ -13,7 +13,8 @@
 class CRUDTaskScheduler(CRUDPlus[TaskScheduler]):
     """任务调度数据库操作类"""
 
-    async def get(self, db: AsyncSession, pk: int) -> TaskScheduler | None:
+    @staticmethod
+    async def get(db: AsyncSession, pk: int) -> TaskScheduler | None:
         """
         获取任务调度
 
@@ -67,7 +68,8 @@ async def create(self, db: AsyncSession, obj: CreateTaskSchedulerParam) -> None:
         :param obj: 创建任务调度参数
         :return:
         """
-        await self.create_model(db, obj)
+        await self.create_model(db, obj, flush=True)
+        TaskScheduler.no_changes = False
 
     async def update(self, db: AsyncSession, pk: int, obj: UpdateTaskSchedulerParam) -> int:
         """
@@ -78,7 +80,11 @@ async def update(self, db: AsyncSession, pk: int, obj: UpdateTaskSchedulerParam)
         :param obj: 更新任务调度参数
         :return:
         """
-        return await self.update_model(db, pk, obj)
+        task_scheduler = await self.get(db, pk)
+        for key, value in obj.model_dump(exclude_unset=True).items():
+            setattr(task_scheduler, key, value)
+        TaskScheduler.no_changes = False
+        return 1
 
     async def set_status(self, db: AsyncSession, pk: int, status: bool) -> int:
         """
@@ -89,7 +95,10 @@ async def set_status(self, db: AsyncSession, pk: int, status: bool) -> int:
         :param status: 状态
         :return:
         """
-        return await self.update_model(db, pk, {'enabled': status})
+        task_scheduler = await self.get(db, pk)
+        setattr(task_scheduler, 'enabled', status)
+        TaskScheduler.no_changes = False
+        return 1
 
     async def delete(self, db: AsyncSession, pk: int) -> int:
         """
@@ -99,7 +108,10 @@ async def delete(self, db: AsyncSession, pk: int) -> int:
         :param pk: 任务调度 ID
         :return:
         """
-        return await self.delete_model(db, pk)
+        task_scheduler = await self.get(db, pk)
+        await db.delete(task_scheduler)
+        TaskScheduler.no_changes = False
+        return 1
 
 
 task_scheduler_dao: CRUDTaskScheduler = CRUDTaskScheduler(TaskScheduler)

backend/app/task/model/scheduler.py

@@ -38,7 +38,6 @@ class TaskScheduler(Base):
     start_time: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), comment='任务开始触发的时间')
     expire_time: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), comment='任务不再触发的截止时间')
     expire_seconds: Mapped[int | None] = mapped_column(comment='任务不再触发的秒数时间差')
-    last_run_time: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), comment='任务最后触发的时间')
     type: Mapped[int] = mapped_column(comment='调度类型（0间隔 1定时）')
     interval_every: Mapped[int | None] = mapped_column(comment='任务再次运行前的间隔周期数')
     interval_period: Mapped[str | None] = mapped_column(String(255), comment='任务运行之间的周期类型')
@@ -58,6 +57,9 @@ class TaskScheduler(Base):
         Boolean().with_variant(INTEGER, 'postgresql'), default=True, comment='是否启用任务'
     )
     total_run_count: Mapped[int] = mapped_column(default=0, comment='任务触发的总次数')
+    last_run_time: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), default=None, comment='任务最后触发的时间'
+    )
     remark: Mapped[str | None] = mapped_column(
         LONGTEXT().with_variant(TEXT, 'postgresql'), default=None, comment='备注'
     )
@@ -66,23 +68,22 @@ class TaskScheduler(Base):
 
     @staticmethod
     def before_insert_or_update(mapper, connection, target):
-        print('before_insert_or_update', mapper, connection, target)
         if target.expire_seconds is not None and target.expire_time:
             raise errors.ConflictError(msg='expires 和 expire_seconds 只能设置一个')
 
     @classmethod
     def changed(cls, mapper, connection, target):
-        print('changed', mapper, connection, target)
         if not target.no_changes:
             cls.update_changed(mapper, connection, target)
 
     @classmethod
-    def update_changed(cls, mapper, connection, target):
-        print('update_changed', mapper, connection, target)
+    async def update_changed_async(cls):
         now = timezone.now()
-        last_update = asyncio.create_task(redis_client.get(f'{settings.CELERY_REDIS_PREFIX}:last_update'))
-        if not last_update:
-            asyncio.create_task(redis_client.set(f'{settings.CELERY_REDIS_PREFIX}:last_update', timezone.to_str(now)))
+        await redis_client.set(f'{settings.CELERY_REDIS_PREFIX}:last_update', timezone.to_str(now))
+
+    @classmethod
+    def update_changed(cls, mapper, connection, target):
+        asyncio.create_task(cls.update_changed_async())
 
 
 # 事件监听器

backend/app/task/schema/scheduler.py

@@ -22,7 +22,6 @@ class TaskSchedulerSchemeBase(SchemaBase):
     start_time: datetime | None = Field(default=None, description='任务开始触发的时间')
     expire_time: datetime | None = Field(default=None, description='任务不再触发的截止时间')
     expire_seconds: int | None = Field(default=None, description='任务不再触发的秒数时间差')
-    last_run_time: datetime | None = Field(default=None, description='任务最后触发的时间')
     type: TaskSchedulerType = Field(default=TaskSchedulerType.INTERVAL, description='任务调度类型（0间隔 1定时）')
     interval_every: int | None = Field(default=None, description='任务再次运行前的间隔周期数')
     interval_period: PeriodType | None = Field(default=None, description='任务运行之间的周期类型')
@@ -32,7 +31,6 @@ class TaskSchedulerSchemeBase(SchemaBase):
     crontab_day_of_month: str | None = Field(default='*', description='运行的每月日期，"*" 表示全部')
     crontab_month_of_year: str | None = Field(default='*', description='运行的月份，"*" 表示全部')
     one_off: bool = Field(default=False, description='是否仅运行一次')
-    total_run_count: int = Field(default=0, description='任务触发的总次数')
     remark: str | None = Field(default=None, description='备注')
 
 

backend/app/task/utils/schedulers.py

@@ -10,7 +10,7 @@
 from celery import current_app, schedules
 from celery.beat import ScheduleEntry, Scheduler
 from celery.utils.log import get_logger
-from sqlalchemy import select, update
+from sqlalchemy import select
 from sqlalchemy.exc import DatabaseError, InterfaceError
 
 from backend.app.task.enums import PeriodType, TaskSchedulerType
@@ -163,8 +163,8 @@ async def from_entry(cls, name, app=None, **entry):
                 task = TaskScheduler(**temp)
                 db.add(task)
             else:
-                stmt = update(TaskScheduler).where(TaskScheduler.name == name).values(**temp)
-                await db.execute(stmt)
+                for key, value in temp.items():
+                    setattr(task, key, value)
             res = cls(task, app=app)
             return res
 
@@ -228,7 +228,7 @@ async def _unpack_fields(
         kwargs: dict | None = None,
         options: dict = None,
         **entry,
-    ):
+    ) -> dict:
         model_schedule = await cls.to_model_schedule(name, task, schedule)
         model_dict = select_as_dict(model_schedule)
         for k in ['id', 'created_time', 'updated_time']:
@@ -254,7 +254,7 @@ def _unpack_options(
         expires: datetime = None,
         expire_seconds: int = None,
         one_off: bool = False,
-    ):
+    ) -> dict:
         data = {
             'queue': queue,
             'exchange': exchange,
@@ -294,7 +294,7 @@ def setup_schedule(self):
         self.install_default_entries(tasks)
         self.update_from_dict(self.app.conf.beat_schedule)
 
-    async def get_all_task_schedules(self):
+    async def get_all_task_schedulers(self):
         """获取所有任务调度"""
         async with async_db_session() as db:
             logger.debug('DatabaseScheduler: Fetching database schedule')
@@ -402,7 +402,7 @@ def schedule(self) -> dict[str, ModelEntry]:
         if update:
             logger.debug('beat: Synchronizing schedule...')
             self.sync()
-            self._schedule = run_await(self.get_all_task_schedules)()
+            self._schedule = run_await(self.get_all_task_schedulers)()
             # 计划已更改，使 Scheduler.tick 中的堆无效
             if not initial:
                 self._heap = []

backend/cli.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 import asyncio
-import os
 
 from dataclasses import dataclass
 from typing import Annotated
@@ -12,12 +11,12 @@
 from rich.panel import Panel
 from rich.text import Text
 from sqlalchemy import text
+from watchfiles import PythonFilter
 
 from backend import console, get_version
 from backend.common.enums import DataBaseType, PrimaryKeyType
 from backend.common.exception.errors import BaseExceptionMixin
 from backend.core.conf import settings
-from backend.core.path_conf import BASE_PATH
 from backend.database.db import async_db_session
 from backend.plugin.tools import get_plugin_sql
 from backend.utils.file_ops import install_git_plugin, install_zip_plugin, parse_sql_script
@@ -45,10 +44,7 @@ def run(host: str, port: int, reload: bool, workers: int | None) -> None:
         address=host,
         port=port,
         reload=not reload,
-        reload_ignore_paths=[
-            os.path.join(BASE_PATH.parent / '.venv'),
-            os.path.join(BASE_PATH / 'log'),
-        ],
+        reload_filter=PythonFilter(),
         workers=workers or 1,
     ).serve()