安裝模塊
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>
<!-- acction指明uri，method指明方法，enctype 必須指明"multipart/form-data" 接收的參數名=表單中的name值 Form表單會返回新的頁面 表單中，每個參數要接一個Form 如果是類型是file，接File -->
- 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)
}
})
})
<!-- ajax 發送文件必須是在FormData對象中 contentType: false, processData: false, 必須都是false才能正確傳輸 -->
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> <!--必須添加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() # 響應內容
前端發送 請求 和接收內容
<!DOCTYPE html>
<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>Document</title>
<style> ul { list-style-type: none; } li { position: relative; height: 40px; } </style>
</head>
<body>
<h1>Websocket</h1>
<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;">aaaaaa</span></li>
<li><span style="float: right;">bbbbbb</span></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