提交任务调度器

requirements.txt

@@ -10,4 +10,5 @@ pydantic==1.10.6
 starlette==0.25.0
 cryptography==39.0.2
 pymongo~=4.3.3
-httpx~=0.23.3
\ No newline at end of file
+httpx~=0.23.3
+APScheduler~=3.10.1
\ No newline at end of file

tools/scheduler.py

+import asyncio
+from apscheduler.jobstores.mongodb import MongoDBJobStore
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.interval import IntervalTrigger
+from pymongo import MongoClient
+
+client = MongoClient('mongodb://localhost:27017/')
+jobstore = MongoDBJobStore(client=client, database='scheduler', collection='jobs')
+
+scheduler = AsyncIOScheduler(jobstores={'mongo': jobstore})
+
+
+async def job_function():
+    await asyncio.sleep(3)
+    print('Hello World')
+
+
+async def blocking_function():
+    while True:
+        print('Blocking')
+        await asyncio.sleep(2)
+
+
+def add_job(job_id, func, trigger=None, args=None, kwargs=None):
+    """添加job"""
+    print(f"添加job - {job_id}")
+    scheduler.add_job(id=job_id, func=func, trigger=trigger, args=args, kwargs=kwargs, jobstore='mongo')
+
+
+def remove_job(job_id):
+    """移除job"""
+    scheduler.remove_job(job_id)
+    print(f"移除job - {job_id}")
+
+
+def pause_job(job_id):
+    """停止job"""
+    scheduler.pause_job(job_id)
+    print(f"停止job - {job_id}")
+
+
+def resume_job(job_id):
+    """恢复job"""
+    scheduler.resume_job(job_id)
+    print(f"恢复job - {job_id}")
+
+
+def get_jobs():
+    """获取所有job信息,包括已停止的"""
+    res = scheduler.get_jobs()
+    print(f"所有job - {res}")
+
+
+def print_jobs():
+    print(f"详细job信息")
+    scheduler.print_jobs()
+
+
+def start():
+    """启动调度器"""
+    scheduler.start()
+
+
+def shutdown():
+    """关闭调度器"""
+    scheduler.shutdown()
+
+
+if __name__ == '__main__':
+    # scheduler.add_job(job_function, trigger=IntervalTrigger(seconds=5), id='my_job', jobstore='mongo')
+    scheduler.start()
+    loop = asyncio.get_event_loop()
+    task = loop.create_task(blocking_function())
+    loop.run_until_complete(asyncio.wait([task]))