FastApi应用和部署生产环境
2021-03-05 19:27
标签：esc VID worker 应用 文件 自己 转发 sql语句 page 最近Python的web框架FastApi火起来了，于是自己也搞了一个，写一下流程。 首先安装fastapi: 新建一个启动文件，就叫main.py吧，引入一下包即可： 运行一下就可以了。输入http://127.0.0.1/docs，就会出现fastapi集成的api文档页面。 因为之前一直写php，习惯了mvc得结构，于是自己大概做了个结构，比如我现在需要一个新闻的模块 video_list.py（相当于控制器） ： 然后在主文件引入 就像插线板一样，将每个小模块路由，插在主文件上。 数据模型层也单独建一个文件，在控制层引入一下： 这里要注意，控制层给模型层传数据时需要验证参数，使用 pydantic，没有的话pip安装一下，好像python 3.6之前的版本不支持这个。 我这里验证的分页参数，具体使用规则，看文档。 这样一个简单的结构就完成了，剩下就是部署到生成环境了。需要nginx做转发： 最后，再做一个守护进程让fastapi一直运行着，这里需要安装一个 gunicorn，当然，如果你有其他方式只要能用就行， 然后在命令行输入： 好了，到此，整个流程就全部完成了。以上都是这两天研究这个框架时简单总结的，如有错误，请留言指出 FastApi应用和部署生产环境 标签：esc VID worker 应用 文件 自己 转发 sql语句 page 原文地址：https://www.cnblogs.com/xubinbk/p/12886633.htmlpip install fastapi
pip install uvicorn
from fastapi import FastAPI
import uvicorn
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
if __name__ == ‘__main__‘:
uvicorn.run(app=‘main:app‘, host="127.0.0.1", port=8000, reload=True, debug=True)from fastapi import APIRouter #引入fastapi的路由
import videoModel #自己的数据模型模块
import verify #自己的验参数模块
router = APIRouter()
@router.post("/video/")
async def video_list(page: verify.PageInfo):
return videoModel.select_videos(page.page, page.pageSize)
from fastapi import FastAPI
import uvicorn
import video_list #引入模块
app = FastAPI()
app.include_router(router=video_list.router) #引入模块路由
@app.get("/")
async def root():
return {"message": "Hello World"}
if __name__ == ‘__main__‘:
uvicorn.run(app=‘main:app‘, host="127.0.0.1", port=8000, reload=True, debug=True)
import pymysql
import config
def select_videos(page, page_size):
conf = config.config()
db = pymysql.connect(host=conf[‘mysql_host‘], port=conf[‘mysql_port‘], user=conf[‘mysql_user‘],
passwd=conf[‘mysql_passwd‘], db=conf[‘mysql_db‘], charset=‘utf8mb4‘)
#unix_socket=‘/Applications/MAMP/tmp/mysql/mysql.sock‘)
result = dict()
limits = (page-1)*page_size
try:
sql = "select videos_id,title,qn_url,qn_thumbnails from `videos` where status = 1 " "order by published_at desc limit "+str(limits)+","+str(page_size)
count_sql = "select count(*) as count from `videos` where status = 1"
# 执行SQL语句
cursor = db.cursor()
cursor.execute(sql)
# 获取所有记录列表
lists = cursor.fetchall()
cursor.execute(count_sql)
# 获取所有记录列表
count = cursor.fetchone()
res = []
for i in lists:
data = dict()
data[‘id‘] = i[0]
data[‘title‘] = i[1]
res.append(data)
result[‘count‘] = count[0]
result[‘list‘] = res
except Exception as e:
print(e)
db.close()
return result
from pydantic import BaseModel
class PageInfo(BaseModel):
page: int = 1
pageSize: int = 10
location / {
proxy_pass http://127.0.0.1:8000/;
}
pip install gunicorn
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker