FastApi_demo_类视图
1.创建app
- 4.1路径参数
- 4.2查询字符串参数
- 4.3表单参数
- 4.4请求体参数
- 4.5依赖 Depends
- 4.6 请求头
安装模块
pip install fastapi1.创建app
from fastapi import FastAPI
app = FastAPI()
@app.get()
async def index():
return ok2.静态文件路径
from fastapi.staticfiles import StaticFiles
static = StaticFiles(directory='./statics') # 静态文件目录名称可以直接写‘statics’
app.mount('/static',static,name='') # 第一个参数是访问静态文件的uri，name的作用是方便反向解析
3.模板(前后端不分离)
from fastapi.templating import Jinja2Templates
template=Jinja2Templates(directory='./templates') # 存放模板的目录
# JinJa2模板引擎 渲染时调用
template.TemplateResponse('模板文件.html',context={'request':request,***})
# 当后端渲染时，必须返回前端request对象，以及其他参数，可以是python对象
# 在html页面模板中，可以使用 {{ python对象 }} 渲染，例如 context={'request':request,'name':'test'}
# 想要渲染出name的值，只需要 {{ name }}，即可在浏览器中显示name对应的值’test‘
# JinJa2模板引擎还提供了for循环、if判断等方法
# example context={'request':request,'a':[1,2,3,4]}
{% for i in a %}
<li>{{ i }}</li>
{% endfor %}
{% for i in a %}
{% if i<2 %}
<li>{{ i }}</li>
{% else %}
<li>{{ i+1 }}</li>
{% endif %}
{% endfor %}
4.参数
4.1路径参数
from fastapi import FastAPI,Path
app = FastAPI()
@app.get('/{id}')
async def func1(id:int): # id就是路径参数，设置了类型校验
pass
@app.get('/test/{id}')
async def func2(id:int=Path(...,ge=1)): # Path是路径参数的类,可以做验证，即对路径参数的校验
pass
Path( # noqa: N802
default: Any, # None为非必传参数，...为必传，指定值为默认
*, # 如果传了*，则说明后面的参数全部为关键字参数
alias: str = None, # 别名，至前端传参时的参数名
title: str = None, # 说明
description: str = None,
gt: float = None, # 只对int/float格式的参数，gt：大于
ge: float = None, # ge ：大于等于
lt: float = None, # lt :小于
le: float = None, # le：小于等于
min_length: int = None, # 只能用于str格式的参数，最小长度
max_length: int = None, # 只能用于str格式的参数，最大长度
regex: str = None, # 只能用于str格式的参数，正则
deprecated: bool = None,
**extra: Any,
)
4.2查询字符串参数
from fastapi import FastAPI,Query
app = FastAPI()
@app.get('/')
async def func1(a:str,b:int): # 直接定义要接收的参数，只能校验类型
pass
@app.get('/1')
async def func2(a:str=Query(),b:int=Query()): # 可以校验类型以及其他的参数，与Path类似
pass
4.3表单参数
from fastapi import Form,UploadFile
from typing import List
app = FastAPI()
@app.post('/index/dc')
async def index_post(request: Request, file1: bytes = File(...), name: str = Form(None), age: int = Form(None)):
print(name)
return name, age
# 或者使用UploadFile进行校验
@app.post('/index/dc')
async def index_post(request: Request, file1: UploadFile = File(...), name: str = Form(None), age: int = Form(None)):
print(name)
return name, age
想要使用表单，必须先安装python-multipart 模块
<form action="./index/dc" method="post" enctype="multipart/form-data">
<input type="file" name="file1">
<input type="text" name="name">
<input type="text" name="age">
<input type="submit" >
form>
- ajax 传输文件
js前端
$('#submit_1').click(function () {
var files = $('#f001').prop('files');
var data = new FormData();
data.append('file0', files[0]);
data.append('age', $("#age0").val());
$.ajax({
url:'./index/aj',
type:'post',
data:data,
contentType: false,
processData: false,
success:function(data) {
alert(data)
}
})
})
python后端
@app.post('/index/aj')
async def index_post(request: Request, file0:bytes=File(...), age: int = Form(None)):
print(file0)
return age
# 接收参数还是Form和File
多文件
多文件只能使用表单方式传输
formdata不能传输列表，如果要实现多文件上传，必须是指定数量的，并与后端一一对应。
<form action="./index/dc" method="post" enctype="multipart/form-data">
<input type="file" name="file1" multiple>
<input type="text" name="name">
<input type="text" name="age">
<input type="submit" >
form>
@app.post('/index/dc')
async def index_post(request: Request,
file0:List[UploadFile]=File(...), # List[UploadFile] 或者
age: int = Form(None)):
print(file0)
return age
4.4请求体参数
- pydantic 类
from pydantic import BaseModel
from typing import Optional
class AAA(BaseModel):
name: Optional[str]
age: Optional[int] = None # int只能用Optional 设置为非必传
- Body
@app.post('/index/pp')
async def index_post(request: Request, aaa: AAA = Body(..., )):
print(aaa.name)
return {'name': 000}
可以不使用Body，默认是JSON格式传输进来的，如果使用了pydantic校验类，默认就是请求体
4.5依赖 Depends
api需要接收的参数是依赖的函数或者类接收的参数，但是接收到的是函数或者类返回的对象
class User(object):
def __init__(self, name, age):
self.name = name
self.age = age
@app.get('/depends')
async def depends_test(request: Request, user: User = Depends(User)):
return user.name, user.age
# 如果依赖的是类，可以直接user: User = Depends()， dastapi会自动识别
def depends_000(name: str, age: int):
return {'name': name, 'age': age}
@app.get('/depends/0')
async def depends_0(request: Request, user: dict = Depends(depends_000)):
return user
# 请求是
# requests.get('http://127.0.0.1:8000/depends/0?name=aaa&age=60')
# 请求需要传的参数是依赖的函数需要的参数，但是在api接口定义时，接收参数校验需要写依赖函数返回值4.6 请求头
请求头与查询字符串一样，定义api时，将需要校验的请求头校验即可
@app.post('/artic/add')
async def add_artic(x_token: str = Header(..., convert_underscores=False)):
return
# convert_underscores 是设置下划线转义的，当请求头中存在下划线，要设置为False
可以结合依赖使用
def check_token(x_token: str = Header(..., convert_underscores=False)):
periods = jwt.decode(x_token, SECRET_KEY, ALGORITHM)
return periods.get('user_id', None)
@app.post('/artic/add')
async def add_artic(request: Request, user_id: str = Depends(check_token),
checktoken: CheckToken = Body(None)):
return
5.响应
- python的类型
from fastapi import FastAPI,Path
app = FastAPI()
@app.get('/')
async def func1():
return '',1,[] # 这里可以直接返回python类型的数据，fastapi会直接转成json
- pydantic的类对象实例
from fastapi import FastAPI,Path
from pydantic import BaseModel
app = FastAPI()
class User(BaseModel):
username:str
age:int
@app.get('/', response_model=User, # 此处的response_model是可选的
response_model_include=[], # 要响应的字段
response_model_exclude=[]) # 不响应的字段，两者二选一
async def func1():
user=User(username='a',age=12)
return user # 直接返回pydantic的BaseModel的子类
- 模板渲染
from fastapi.templating import Jinja2Templates
app = FastAPI()
template=Jinja2Templates(directory='./templates').
@app.get('/')
async def func1():
return template.TemplateResponse('模板文件.html',context={'request':request,***})
- 纯文本
from fastapi.templating import Jinja2Templates
app = FastAPI()
@app.get('/', response_model=User)
async def func1():
return HTMLResponse('html格式的数据')
- 特殊文件
@app.get('/img')
async def get_img(request: Request):
return FileResponse(
'./statics/img/000.jpg',
media_type='image/jpg')
6.中间件
- 装饰器方法
@app.middleware('http')
def my_middleware(request:Request, call_next):
# 这里进行访问视图前的处理
# 在这里给request添加属性是没有意义的，不会传递下去 具体原因还在找
response = call_next(request)
# 这里进行访问视图后的处理
return response
7.CORS跨域请求
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
origins = [
"http://localhost.tiangolo.com",
"https://localhost.tiangolo.com",
"http://localhost",
"http://localhost:8080",
]
app.add_middleware(
CORSMiddleware,
allow_origins=origins, # 应该允许进行跨域请求的来源列表
allow_credentials=True, # 表示跨域请求应支持cookie
allow_methods=["*"], # 指出应使浏览器可以访问的所有响应标头
allow_headers=["*"], # 跨域请求应允许的HTTP方法列表
)
8.异常处理Haddler
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):
if item_id not in items:
# 需要抛出异常时，直接抛出HTTPException，会自动转换成json格式返回给前端
raise HTTPException(
status_code=404,
detail="Item not found",
headers={"X-Error": "There goes my error"},
)
return {"item": items[item_id]}
# 也可以自定义异常类
class UnicornException(Exception):
def __init__(self, name: str):
self.name = name
@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
return JSONResponse(
status_code=418,
content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
)
# 当需要抛出异常时，直接抛出自定义的异常类
9.BackgroundTasks
后台任务，当需要指定的任务处理较慢，可以放在后台处理，可以先返回202（已接收的状态码）并在后台进行处理
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_notification(email: str, message=""):
with open("log.txt", mode="w") as email_file:
content = f"notification for {email}: {message}"
email_file.write(content)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "Notification sent in the background"}
10.类视图
fastapi本身是没有类视图的概念的，实现类视图要借助 starlette.endpoints 下的 HTTPEndpoint或者WebSocketEndpoint
- HTTPEndpoint 针对http请求
from fastapi import FastAPI
from starlette.endpoints import HTTPEndpoint, WebSocketEndpoint
app = FastAPI()
class Test_HTTP(HTTPEndpoint):
async def get(self, request: Request):
return ''
async def post(self, '其他参数'):
return ''
app.add_route('/test',Test_HTTP,name='aaa')
# 这种方式添加的视图，没有docs和redoc文档
app.add_api_route('/test2',func,name='bbb') # 这个方法只能添加函数视图，因为底层是用fastapi的APIRoute创建的，只支持函数，也没有docs/redoc文档
- WebSocketEndpoint 针对websocket
11.JWT
FastAPI推荐使用JWT进行校验
from jose import JWTError, jwt
# jwt加密
# data:要加载的内容
expire = datetime.utcnow() + timedelta(minutes=expires_delta) # expires_delta ： int
data.update({'exp': expire})
jwt.encode(data, SECRET_KEY, algorithm=ALGORITHM) # 生成的就是jwt_token
# jwt解密
jwt.decode(data, SECRET_KEY, algorithm=ALGORITHM) # 加密后就是原来的字典
# 当签名过期时会报500错误
from passlib.context import CryptContext
# hash加密密码
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# 加密
hashed_password =pwd_context.hash(password)
# 校验
pwd_context.verify(password,hashed_password) # 返回Bool值
# 模拟用户
user_db = {
'ID': 'TMXC_001',
'username': 'TMXC_admin',
'password': pwd_context.hash('TMXC_password')
}
# 设置加密码
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
# 生成token
def create_token(data: dict, expires_delta: Optional[int] = None):
if expires_delta:
expire = datetime.utcnow() + timedelta(minutes=expires_delta)
else:
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
data.update({'exp': expire})
access_token = jwt.encode(data, SECRET_KEY, algorithm=ALGORITHM)
return access_token
# 发送请求响应token
@app.post('/token')
async def get_token(request: Request, form_data: OAuth2PasswordRequestForm = Depends()):
username = form_data.username
password = form_data.password
print(username, password)
if username == user_db.get('username') and pwd_context.verify(password, user_db.get('password')):
data = {'user_id': user_db.get('ID')}
else:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
detail='用户名或密码错误')
token_code = create_token(data=data, expires_delta=3)
# token = Tokenizer(token_code)
# 前后端分离的可以直接传入，前端接收即可
return token_code
# 当前后端不分离，跳转新页面时，将token按传入，前端用jinja2模板引擎的方式接收
# return template.TemplateResponse('add_art.html', context={'request': request, 'token_dict': token})
# 校验token
# 校验时，当过期会报错
def check_token(x_token: str = Header(..., convert_underscores=False)):
try:
periods = jwt.decode(x_token, SECRET_KEY, ALGORITHM)
print(datetime.utcfromtimestamp(periods.get('exp')))
except ExpiredSignatureError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail='token已过期')
return periods.get('user_id', None)
# 需要密钥才能访问的页面，利用依赖和请求头进行校验
@app.post('/artic/add')
async def add_artic(request: Request, user_id: str = Depends(check_token),
checktoken: CheckToken = Body(None)):
# user_id = check_token(x_token)
print(datetime.utcnow())
if user_id:
return '11111111'
html
var token = '';
$("#btn1").click(
function () {
$.ajax({
type: "POST",
url:'./artic/add',
headers: {'x_token':token}, // 此处添加请求头
data: JSON.stringify({'title':$('#tt1').val(),'count':$("#cnt").val()}),
success:function (data,status){
alert(data)
},
error:function (data){
alert(data.detail)
}
})
});
12 WebSocket
后端搭建websocket服务器
from fastapi import WebSocket
@app.websocket('/ws')
async def websocket_dep(websocket:WebSocket):
await websocket.accept()
while True:
data =await websocket.receive_text()
websocket.send_text() # 响应内容
前端发送 请求 和接收内容
<html lang="en">
<head>
<meta charset="UTF-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Documenttitle>
<style>
ul {
list-style-type: none;
}
li {
position: relative;
height: 40px;
}
style>
head>
<body>
<h1>Websocketh1>
<form action="">
<input type="text" name="mess" id="msg">
<input type="button" value="submit" id="btn">
form>
<ul id='rst'>
<li><span style="float: left;">aaaaaaspan>li>
<li><span style="float: right;">bbbbbbspan>li>
ul>
<script>
var ws = new WebSocket('ws://localhost:8000/ws');
ws.onmessage = function(event){
li = document.createElement('li');
span = document.createElement('span');
span.innerHTML = event.data; // 接收消息
span.style.float = 'left';
li.appendChild(span);
rst = document.getElementById('rst');
rst.appendChild(li);
}
var msg = document.getElementById('msg');
var btn = document.getElementById('btn');
btn.onclick = function(event){
ws.send(msg.value) // 发送消息
li = document.createElement('li');
span = document.createElement('span');
span.innerHTML = msg.value;
span.style.float = 'right';
li.appendChild(span);
msg.value = '';
rst = document.getElementById('rst');
rst.appendChild(li);
// event.preventDefault(); // 不跳转
return false // 不跳转
}
script>
body>
html>
托管
from fastapi import WebSocket
# 用于管理websocket的连接
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def send_json(self,data,websocket: WebSocket):
await websocket.send_json(data)
# 给所有连接发送文本
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.send_message(f"You wrote: {data}", websocket)
await manager.broadcast(f"Client #{client_id} says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(f"Client #{client_id} left the chat")
13.部署 uvicorn + supervisor
uvicorn 运行server后，退出当前终端，项目进程会自动被杀死，可以使用supervisor 进行管理
# 安装 supervisor
sudo apt-get install supervisor
# 配置文件
sudo vi /etc/supervisor/conf.d/app.conf
# 写入一下内容
[program:tmxc]
directory = /home/ubuntu/tmxc/ ; 程序的启动目录
command = /home/ubuntu/.virtualenvs/tmxc/bin/uvicorn main:app --reload ; 启动命令，与命令行启动的命令是一样的
autostart = true ; 在 supervisord 启动的时候也自动启动
startsecs = 5 ; 启动 5 秒后没有异常退出，就当作已经正常启动了
autorestart = true ; 程序异常退出后自动重启
startretries = 3 ; 启动失败自动重试次数，默认是 3
user = ubuntu ; 用哪个用户启动
redirect_stderr = true ; 把 stderr 重定向到 stdout，默认
# 重启supervisor**** 重要
sudo /etc/init.d/supervisor restart
# 重启之后已经运行起来了，当更新了配置文件后，需要重启
sudo supervisorctl restart tmxc