有一個家用需求：
以rest api的形式釋出一個服務（函式）
訪問這個服務需要驗證，確切說是token（JWT）認證
訪問這個服務的request/response資料需要被校驗
自帶如何使用這個api的文件
簡單點
官方文件詳細豐富
效能最不重要
python實現的
就功能實現上，任何一個主流web後端框架都具備，但自動生成api使用文件，對資料校驗方便直觀，整體使用簡單等，FastAPI就具備這些。
FastAPI是一個基於python的web後端框架，使用起來開發效率高，更少的程式碼量；開發完畢後，會
自動生成API使用文件
，扔給對方就可以了；基於pydantic，方便的對
資料校驗
；更高的效能，
支援ASGI規範
，也就是支援非同步，
支援WebSocket
。
等等還有一些其他web後端框架，例如
Django： 重型web框架，功能強大，官方文件豐富，很適合開發正兒八經的web應用
Flask：中型web框架，web應用，rest api用途，都可以
Bottle： 微型web框架，用來做個rest api用途，再合適不過了，這個框架本身就一個單檔案，沒有任何其他依賴，也就是說在安裝了python直譯器的環境，把這個單檔案複製過來就可以使用了，在一些Network OS裡面，就是使用的這個框架對外提供簡單的服務。
下面給出在官方的demo的基礎上完善的程式碼，有需要拿過去再改改就可以跑生產：
from
datetime
import
datetime
，
timedelta
from
typing
import
Optional
from
fastapi
import
Depends
，
FastAPI
，
HTTPException
，
status
，
Body
from
fastapi。security
import
OAuth2PasswordBearer
，
OAuth2PasswordRequestForm
from
jose
import
JWTError
，
jwt
from
passlib。context
import
CryptContext
from
pydantic
import
BaseModel
# https：//fastapi。tiangolo。com/tutorial/security/oauth2-jwt/
# to get a string like this run in linux：
# openssl rand -hex 32
SECRET_KEY
=
“c4af8692b37bcf2d575c5958254eee21a049cf01925207beb8e4f02a5c0c9593”
ALGORITHM
=
“HS256”
ACCESS_TOKEN_EXPIRE_MINUTES
=
30
# secret01
fake_users_db
=
{
“johndoe”
：
{
“username”
：
“johndoe”
，
“full_name”
：
“John Doe”
，
“email”
：
“johndoe@example。com”
，
“hashed_password”
：
“$2b$12$X8w2pubd67dP8JsAijamkejzK1LIiY。JMTh7qAgscb9TVkSHkT0sy”
，
“disabled”
：
False
，
}
}
class
Test01_data_model
（
BaseModel
）：
infor
：
Optional
［
str
］
=
None
class
Test02_data_model
（
BaseModel
）：
username
：
str
MFAtoken
：
str
addtion
：
Optional
［
str
］
=
None
class
Token
（
BaseModel
）：
access_token
：
str
token_type
：
str
class
TokenData
（
BaseModel
）：
username
：
Optional
［
str
］
=
None
class
User
（
BaseModel
）：
username
：
str
：
Optional
［
str
］
=
None
full_name
：
Optional
［
str
］
=
None
disabled
：
Optional
［
bool
］
=
None
class
UserInDB
（
User
）：
hashed_password
：
str
pwd_context
=
CryptContext
（
schemes
=
［
“bcrypt”
］，
deprecated
=
“auto”
）
# state this url-“\token” is used for get token only
oauth2_scheme
=
OAuth2PasswordBearer
（
tokenUrl
=
“token”
）
app
=
FastAPI
（）
def
verify_password
（
plain_password
，
hashed_password
）：
return
pwd_context
。
verify
（
plain_password
，
hashed_password
）
# used for generating hash_password by plain_password
def
get_password_hash
（
password
）：
return
pwd_context
。
hash
（
password
）
def
get_user
（
db
，
username
：
str
）：
if
username
in
db
：
user_dict
=
db
［
username
］
return
UserInDB
（
**
user_dict
）
def
authenticate_user
（
fake_db
，
username
：
str
，
password
：
str
）：
user
=
get_user
（
fake_db
，
username
）
if
not
user
：
return
False
if
not
verify_password
（
password
，
user
。
hashed_password
）：
# the former password is a plain，yes！！
return
False
return
user
def
create_access_token
（
data
：
dict
，
expires_delta
：
Optional
［
timedelta
］
=
None
）：
to_encode
=
data
。
copy
（）
if
expires_delta
：
expire
=
datetime
。
utcnow
（）
+
expires_delta
else
：
expire
=
datetime
。
utcnow
（）
+
timedelta
（
minutes
=
15
）
to_encode
。
update
（{
“exp”
：
expire
}）
encoded_jwt
=
jwt
。
encode
（
to_encode
，
SECRET_KEY
，
algorithm
=
ALGORITHM
）
return
encoded_jwt
async
def
get_current_user
（
token
：
str
=
Depends
（
oauth2_scheme
））：
‘’‘
this function is used for verifying the token’s expire time and get the user‘s infor
’‘’
（
token
）
credentials_exception
=
HTTPException
（
status_code
=
status
。
HTTP_401_UNAUTHORIZED
，
detail
=
“Could not validate credentials”
，
headers
=
{
“WWW-Authenticate”
：
“Bearer”
}，
）
try
：
payload
=
jwt
。
decode
（
token
，
SECRET_KEY
，
algorithms
=
［
ALGORITHM
］）
‘’‘
# A token can be decode to be as the data was encoded before
if the token has expired then it can not be decode and raise a
’ExpiredSignatureError（‘Signature has expired。’）‘ error。
payload = {’sub‘： ’johndoe‘， ’exp‘： 1634789405}
’‘’
（
payload
）
username
：
str
=
payload
。
get
（
“sub”
）
if
username
is
None
：
raise
credentials_exception
token_data
=
TokenData
（
username
=
username
）
except
Exception
as
e
：
（
repr
（
e
））
raise
credentials_exception
user
=
get_user
（
fake_users_db
，
username
=
token_data
。
username
）
if
user
is
None
：
raise
credentials_exception
return
user
async
def
get_current_active_user
（
current_user
：
User
=
Depends
（
get_current_user
））：
if
current_user
。
disabled
：
raise
HTTPException
（
status_code
=
400
，
detail
=
“Inactive user”
）
return
current_user
@app。post
（
“/token”
，
response_model
=
Token
）
async
def
login_for_access_token
（
form_data
：
OAuth2PasswordRequestForm
=
Depends
（））：
user
=
authenticate_user
（
fake_users_db
，
form_data
。
username
，
form_data
。
password
）
if
not
user
：
raise
HTTPException
（
status_code
=
status
。
HTTP_401_UNAUTHORIZED
，
detail
=
“Incorrect username or password”
，
headers
=
{
“WWW-Authenticate”
：
“Bearer”
}，
）
access_token_expires
=
timedelta
（
minutes
=
ACCESS_TOKEN_EXPIRE_MINUTES
）
access_token
=
create_access_token
（
data
=
{
“sub”
：
user
。
username
}，
expires_delta
=
access_token_expires
）
return
{
“access_token”
：
access_token
，
“token_type”
：
“bearer”
}
@app。get
（
“/users/me/”
，
response_model
=
User
）
async
def
read_users_me
（
current_user
：
User
=
Depends
（
get_current_active_user
））：
return
current_user
@app。get
（
“/users/me/items/”
）
async
def
read_own_items
（
current_user
：
User
=
Depends
（
get_current_active_user
））：
return
［{
“item_id”
：
“Foo”
，
“owner”
：
current_user
。
username
}］
@app。get
（
“/test01/”
，
response_model
=
Test01_data_model
，
dependencies
=
［
Depends
（
get_current_active_user
）］）
async
def
test01_app
（）：
return
{
“infor”
：
“tokenttttttttt”
}
@app。post
（
“/test02/”
，
dependencies
=
［
Depends
（
get_current_active_user
）］）
async
def
test02_app
（
user_infor
：
Test02_data_model
=
Body
（
。。。
））：
（
user_infor
。
dict
（））
return
{
“infor”
：
user_infor
}
if
__name__
==
‘__main__’
：
pass
#import uvicorn
#uvicorn。run（app=‘fastapi_token_demo：app’， host=“127。0。0。1”， port=8000， reload=True， debug=True）
#python -m uvicorn fastapi_token_demo：app ——host ‘127。0。0。1’ ——port 8000 ——reload
一些說明如下:
關於Token認證的流程：
當去訪問一些服務時，如果直接使用使用者憑據，當憑據洩露時，攻擊者便可以登入你主頁，擁有全部控制權，token認證是指先用使用者憑據認證一次，獲取到一個字串，其中被
編碼了使用者資訊sub
和該字串
超時資訊exp
，後續訪問服務時，只需攜帶token即可，即使token洩露，影響是可控的，不同服務類別，獲取不同的token，並且token超時會自動失效，這個也是目前對API類應用訪問的主流認證方式。
需要安裝的庫
pip install fastapi
pip install “uvicorn［standard］” //work as the ASGI server that runs your code
pip install python-multipart //get values from html form data
pip install “python-jose［cryptography］” //generate and verify the JWT tokens
pip install “passlib［bcrypt］” //handle password hashes
生成一個隨機字串，SECRET_KEY ，用於後續JWT的encode 和decode
openssl rand -hex 32
關於 ACCESS_TOKEN_EXPIRE_MINUTES ，由於JWT沒有回收機制，每次生成token後，只有等待超時才能過期，實際使用許謹慎設定過長的token超期時間，當然可以更改SECRET_KEY強制所有token報廢。
fake_users_db ，實際使用可以從資料庫select使用者表資訊
關於使用者的hashed_password，可以由下面定義的get_password_hash 函式生產
get_current_user函式中，呼叫了jwt。decode方法，用於解碼JWT字串，如果不是由同一個例項，jwt。encode方法產生的，或者exp超時，會raise錯誤，這個decode的過程實際是完成了對JWT的認證過程。
關於oauth2_scheme = OAuth2PasswordBearer（tokenUrl=“token”），當被呼叫時，會檢查request的
Authorization
header，如果該頭部值是Bearer+str（token），會取出該值，如果不符合條件，會向客戶端返回404error 和 UNAUTHORIZED資訊， oauth2_scheme是包含token的字串值，可以被其他需要JWT的函式Depends。
fastapi中引入Depends（），讓函式之間的依賴更直觀，邏輯清晰；當A函式被宣告依賴另一個B函式，A執行時，會先執行B，只有當B成功執行，為True時，A函式才會繼續，那麼在所有需要token認證的函式下，宣告依賴get_current_active_user即可，也即是依賴get_current_user函式的jwt。decode方法。
dependencies=［B，C］ 用於宣告多個依賴的語法
關於async， 如果你定義的函式下有IO的操作，並且你
不清楚是否支援非同步，請不要用async關鍵字
，沒有async，fastapi會自動判斷是否函式阻塞，如果阻塞，會呼叫執行緒池，但是如果在async函式下，引入了不支援非同步的第三方IO庫，妥妥變成同步！！
關於request 傳參，如果同時存在 fastapi會依據以下特點判斷對應關係
路徑引數：url路徑，函式透過 引數名字識別，對應
查詢引數：例如get方法，宣告為
singular type（
int
，
float
，
str
，
bool
）
的都為查詢引數
，
request Body： 對該引數定義了
Pydantic model
的，為Body資料，送到事先定義好的資料模型校驗和轉換，轉成json後賦值給引數，不滿足資料模型校驗會返回客戶端錯誤
from
typing
import
Optional
from
fastapi
import
FastAPI
from
pydantic
import
BaseModel
class
Item
（
BaseModel
）：
name
：
str
description
：
Optional
［
str
］
=
None
price
：
float
tax
：
Optional
［
float
］
=
None
app
=
FastAPI
（）
@app。put
（
“/items/{item_id}”
）
async
def
create_item
（
item_id
：
int
，
item
：
Item
，
q
：
Optional
［
str
］
=
None
）：
result
=
{
“item_id”
：
item_id
，
**
item
。
dict
（）}
if
q
：
result
。
update
（{
“q”
：
q
}）
return
result
再給個例子：
from
typing
import
Optional
from
fastapi
import
Body
，
FastAPI
from
pydantic
import
BaseModel
app
=
FastAPI
（）
class
Item
（
BaseModel
）：
name
：
str
description
：
Optional
［
str
］
=
None
price
：
float
tax
：
Optional
［
float
］
=
None
class
User
（
BaseModel
）：
username
：
str
full_name
：
Optional
［
str
］
=
None
@app。put
（
“/items/{item_id}”
）
async
def
update_item
（
item_id
：
int
，
item
：
Item
，
user
：
User
，
importance
：
int
=
Body
（
。。。
）
）：
results
=
{
“item_id”
：
item_id
，
“item”
：
item
，
“user”
：
user
，
“importance”
：
importance
}
return
results
需要傳入的json格式
{
“item”： {
“name”： “Foo”，
“description”： “The pretender”，
“price”： 42。0，
“tax”： 3。2
}，
“user”： {
“username”： “dave”，
“full_name”： “Dave Grohl”
}，
“importance”： 5
}
部署+測試
python -m uvicorn fastapi_token_demo：app ——host ‘127。0。0。1’ ——port 8000 ——reload
啟動後，會自動生成API相關文件
http：//
x。x。x。x：8000/docs
這裡使用用Talend（postman）測試
向
http：//
x。x。x。x/token
發post請求，表單資料為
username：johndoe
password：secret01 //取決你實際的hashed_password的明文
認證成功後會返回你一個json，如下：
Talend（postman）工具
在後續的request中新增HTTP 認證頭部，帶上JTW
Authorization ： Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9。eyJzdWIiOiJqb2huZG9lIiwiZXhwIjoxNjM3Mzk4MTEwfQ。apcFjARS2StEnOum3fCJOKWZ02vvaNzE-tsOQgz0pMU
透過JWT認證如下，get
http：//
x。x。x。x/users/me/
one more things:
這部分談談完整的部署架構
web後端框架——-WSGI/ASGI————-Ngnix/Apache————Brower
這裡區分下Ngnix/Apache 和uWSGI（Web Server Gateway Interface）的作用：
web框架只負責處理各種業務邏輯，uWSGI負責接收client的網路請求
，呼叫（函式呼叫）web框架處理請求，接收web框架的資料響應，返回資料給client，uWSGI和web框架間按照一定規則呼叫和被呼叫，共同完成對client的請求，這個規則的集合就是WSGI，具體的實現之一是uWSGI，由於標準化，
uWSGI可以對接多種web框架
ASGI（
Asynchronous Server Gateway Interface
）是對WSGI標準的擴充套件，包含WSGI的全部特性，新增加了，websocket協議，非同步 的支援，具體的實現有
Uvicorn，
那麼好像對於一個完整的訪問，沒有Ngnix/Apache這些元件的什麼事，確實是！
在實際生產部署中，Ngnix/Apache這類的元件可以算是對uWSGI和
Uvicorn一個助力，
例如 對uWSGI和Uvicorn server的負載；接管來自client的請求中包含的靜態資源的訪問；實現SSL offload；對Client請求的URL的攔截等等，其他‘動態’的資源，計算處理啊，資料庫查詢等請求會轉發到uWSGI和Uvicorn server。
fastapi高效能部署架構：
對於WSGI web框架，以上就是實際的部署架構了，但是對於ASGI web框架，如果要充分發揮高效能，高併發，就有一些改變了，GitHub有一個專案，把該架構整體打包，作為一個docker映象釋出，方便大家使用
fastapi高效能部署架構
fastapi——ASGI————-WSGI——-Ngnix/Apache————Brower
增加一層WSGI元件，使用的是Gunicorn，另外一個WSGI的實現，ASGI的實現是使用的
Uvicorn，
Gunicorn的作用是當作一個程序管理器，管理多個
Uvicorn程序，提高併發，
說起來很複雜，但藉助
tiangolo/uvicorn-gunicorn-fastapi-docker 這個開源專案，部署和使用起來都非常簡單
來試試看，假設你的業務程式碼已經準備好，放在本地app資料夾下
官方GitHub上提供了基於各個python版本的docker映象，這次我選擇uvicorn+gunicorn+fastapi+python3。8 docker映象，剩下的只需要在把app資料夾和程式碼需要的第三方python庫加入，重新docker build即可
假如我用fastapi釋出了一個基於selenium庫的web自動化功能，並且這個功能需要fastapi token認證，那麼我的
requirements.txt 檔案如下：
selenium==3。141。0
python-jose［cryptography］==3。3。0
passlib［bcrypt］==1。7。4
python-multipart==0。0。5
Dockerfile檔案如下：
FROM tiangolo/uvicorn-gunicorn-fastapi：python3。8
LABEL maintainer=“Sebastian Ramirez
COPY requirements。txt /tmp/requirements。txt
RUN pip install ——no-cache-dir -r /tmp/requirements。txt
COPY 。/app /app
我的檔案目錄如下：
。
|—— app
| |—— main。py
| |—— selenium_timesheet。py
| `—— stealth。min。js
|—— Dockerfile
|—— README。md
`—— requirements。txt
務必把fastapi主檔案更改為main。py
docker build -t myimage 。
docker run -d ——name mycontainer_fastapi -p 8000：80 myimage
以上。
