中文文档:FastAPI
API接口开发其实特简单,Python FastApi Web 框架教程来了_根据接口文档用fastapi开发接口-CSDN博客
Python框架篇(1):FastApi-快速入门 - 知乎
.
├── app # 「app」是一个 Python 包
│ ├── __init__.py # 这个文件使「app」成为一个 Python 包
│ ├── main.py # 「main」模块,例如 import app.main
│ ├── dependencies.py # 「dependencies」模块,例如 import app.dependencies
│ └── routers # 「routers」是一个「Python 子包」
│ │ ├── __init__.py # 使「routers」成为一个「Python 子包」
│ │ ├── items.py # 「items」子模块,例如 import app.routers.items
│ │ └── users.py # 「users」子模块,例如 import app.routers.users
│ └── internal # 「internal」是一个「Python 子包」
│ ├── __init__.py # 使「internal」成为一个「Python 子包」
│ └── admin.py # 「admin」子模块,例如 import app.internal.admin
????????main.py
import time
from typing import Union
from apscheduler.schedulers.background import BackgroundScheduler
from fastapi import FastAPI
import uvicorn
from contextlib import asynccontextmanager
'''
启动命令
uvicorn main:app --reload --port 8000
#导出依赖
pip freeze >requirements.txt
api文档地址
http://localhost:8080/docs
'''
data_task = BackgroundScheduler()
task_id = 'data_task_1'
@asynccontextmanager
async def lifespan(app: FastAPI):
# 星期2——6 5:30 执行一次
# scheduler.add_job(func=task_function, trigger="cron", day_of_week='1,2,3,4,6', hour=5,minute=30)
data_task.add_job(func=task_function, trigger="interval", seconds=1,id=task_id)
data_task.start()
# print("启动前执行")
yield
# print("关闭后前执行")
app = FastAPI(lifespan=lifespan)
def task_function():
print(f"Current Time: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())}")
@app.get("/scheduler/stop")
def stop():
data_task.pause_job(task_id)
print(task_id + ' is stop')
return task_id + ' is stop'
@app.get("/scheduler/start")
def start():
data_task.resume_job(task_id)
print(task_id + ' is start')
return task_id + ' is start'
# 第二种启动方式:
if __name__ == '__main__':
uvicorn.run(app="main:app", host="localhost", port=8000)
参考 :
APScheduler定时任务框架 - 星空看海 - 博客园
crontab 定时任务、Python 定时任务框架 apscheduler_crontab 定时 执行 python 脚本-CSDN博客