fastapi 介紹

FastAPI 是一個(gè)用于構(gòu)建 API 的現(xiàn)代、快速（高性能）的 web 框架，使用 Python 3.6+ 并基于標(biāo)準(zhǔn)的 Python 類型提示。
個(gè)人比較喜歡自帶的swagger

官方中文文檔

關(guān)鍵特性:

快速：可與 NodeJS 和 Go 比肩的極高性能（歸功于 Starlette 和 Pydantic）。最快的 Python web 框架之一。
高效編碼：提高功能開發(fā)速度約 200％ 至 300％。*
更少 bug：減少約 40％ 的人為（開發(fā)者）導(dǎo)致錯(cuò)誤。*
智能：極佳的編輯器支持。處處皆可自動補(bǔ)全，減少調(diào)試時(shí)間。
簡單：設(shè)計(jì)的易于使用和學(xué)習(xí)，閱讀文檔的時(shí)間更短。
簡短：使代碼重復(fù)最小化。通過不同的參數(shù)聲明實(shí)現(xiàn)豐富功能。bug 更少。
健壯：生產(chǎn)可用級別的代碼。還有自動生成的交互式文檔。
標(biāo)準(zhǔn)化：基于（并完全兼容）API 的相關(guān)開放標(biāo)準(zhǔn)：OpenAPI (以前被稱為 Swagger) 和 JSON Schema。

fastapi安裝

1.安裝fastapi: pip install fastapi
2.如果用于生產(chǎn)，那么你還需要一個(gè)ASGI服務(wù)器，如Uvicorn或Hypercorn: pip install uvicorn

介紹下基礎(chǔ)代碼結(jié)構(gòu)

在原有的基礎(chǔ)上添加baseurl，目的是用代理時(shí)更好的區(qū)分不同服務(wù)

baseurl = '/xxxx'

app = FastAPI(
    title="WB", #swagger 標(biāo)題
    version="1.0", #版本
    description="API介紹", #描述
    docs_url= baseurl+'/docs', #swagger url
    redoc_url=baseurl+'/redoc',# redoc url
    openapi_url=baseurl+'/openapi.json', # OpenAPI url
    swagger_ui_oauth2_redirect_url=baseurl+"/docs/oauth2-redirect"
)

允許跨域資源共享

app.add_middleware(
    CORSMiddleware,
    # allow_origins=origins,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

用戶登錄

@app.post(baseurl+"/token", response_model=Token,summary='登錄',tags=["用戶登錄"])
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
    )
    logger.info({"access_token": access_token, "token_type": "bearer"})
    return {"access_token": access_token, "token_type": "bearer"}

生成token

def create_access_token(data: dict, expires_delta: Union[timedelta, None] = 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

認(rèn)證token

async def get_current_user(token: str = Depends(oauth2_scheme)):
    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])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        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(baseurl+"/event/uploadfile/",summary='上傳文件',tags=["事件"])
async def create_upload_files(Path: str,files: List[UploadFile],current_user: User = Depends(get_current_active_user)):
    get_timestamp_uuid = uuid.uuid1()  # 根據(jù) 時(shí)間戳生成 uuid , 保證全球唯一
    ID_event = str(get_timestamp_uuid)
    message = {
        'msg':'In the processing'
    }

    for file in files:
        # print(file.filename)
        contents = await file.read()
        filepath = os.path.dirname(os.path.realpath(sys.argv[0]))+f"/files/{ID_event}/" + file.filename
        if not os.path.exists(os.path.split(filepath)[0]):
            # 目錄不存在創(chuàng)建，makedirs可以創(chuàng)建多級目錄
            os.makedirs(os.path.split(filepath)[0])
            # print('os.makedirs')
        with open(filepath, "wb") as f:
            # 2.3 將獲取的fileb文件內(nèi)容，寫入到新文件中
            f.write(contents)
      
    message = {
        'msg': 'success'
    }
    event_lib[ID_event] = message
    return {"ID_event": ID_event,'msg':message[msg]}

查詢結(jié)果

@app.get(baseurl+"/event/Results/",summary='獲取事件結(jié)果',tags=["事件"])
async def Results(ID_event: str ,current_user: User = Depends(get_current_active_user)):
    # print(event_lib)
    if ID_event not in event_lib.keys():
        logger.info({"ID_event": ID_event,'Result':'Event does not exist', "Owner": current_user.username})
        return {"ID_event": ID_event,'Result':'Event does not exist', "Owner": current_user.username}
    logger.info({"ID_event": ID_event,'Result':event_lib[ID_event], "Owner": current_user.username})
    return {"ID_event": ID_event,'Result':event_lib[ID_event], "Owner": current_user.username}

運(yùn)行

if __name__ == '__main__':
    # uvicorn.run(app)
    uvicorn.run(app, host="0.0.0.0",port=8000)