标签：异步 web fastapi app item notification FastAPI def
1 如何快速实现一个fastapi
1.1 准备工作
使用Python来写后端，基本上使用的是Django和Flask。fastapi不仅仅高效率而且还很适合产品级的开发。
需要安装一下fastapi以及ASGI（ASGI是WSGI的升级版，支持异步调用）
$ pip install fastapi
$ pip install pydantic
$ pip install uvicorn
$ pip install python-multipart
1.2 实现一个demo
- 创建
main.py, 然后写入一下代码，可以直接运行
# -*- coding: utf-8 -*-
from fastapi import FastAPI
import uvicorn
# from web_server.views import router
app = FastAPI()
@app.get("/", tags=['首页API'], description='fastapi后台测试服务首页API')
def index():
return {'message': '脚本后台服务已启动。', 'docs_url': 'http://127.0.0.1:8000/docs'}
@app.get("/items/{item_id}")
def read_item(item_id: int, keyword: str):
"""
测试api
:param item_id: url路径参数
:param keyword: url查询字符串参数
:return:
"""
return {"item_id": item_id, "keyword": keyword}
# 添加子路由扩展API
# app.include_router(router, prefix='/exec_script')
if __name__ == '__main__':
print('已启动服务')
# main指的是当前py文件名，app指的是定义的FastAPI()实例
uvicorn.run(app='main:app', host='127.0.0.1', port=8000, reload=True, debug=True)
- 因为安装了
uvicorn库， 还可以通过终端命令运行：
$ uvicorn main:app --reload --port 8000
$ curl http://127.0.0.1:8000
$ curl http://127.0.0.1:8000/items/1?keyword=book
- 响应的结果数据如下：
{
"item_id": 1,
"keyword": "book"
}
2 fastapi后台异步任务
2.1 后台任务的使用
-
创建一个异步任务函数
from fastapi import BackgroundTasks, FastAPI app = FastAPI() def write_notification(email: str, message=""): with open("log.txt", mode="w") as f: f.write(f"notification for {email}:{message}") -
调用后台任务
@app.post("/send-notification/{email}") async def send_notification(email: str, background_tasks: BackgroundTasks): background_tasks.add_task(write_notification, email, message="notification...") return {"message": "Notification sent in the background"}-
add_task()接收的参数：
-
一个在后台运行的任务函数（write_notification）
-
按照顺寻传递的一系列参数（email）
-
任何的关键字参数（message="notification..."）
-
-
2.2 依赖注入
from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
return commons
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
return commons
标签：异步,web,fastapi,app,item,notification,FastAPI,def 来源： https://www.cnblogs.com/itelephant/p/15948766.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。