python fastapi 强大简单的后端框架

fastapi 介绍

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 web 框架，使用 Python 3.6+ 并基于标准的 Python 类型提示。
个人比较喜欢自带的swagger

官方中文文档

关键特性:

快速：可与 NodeJS 和 Go 比肩的极高性能（归功于 Starlette 和 Pydantic）。最快的 Python web 框架之一。
高效编码：提高功能开发速度约 200％ 至 300％。*
更少 bug：减少约 40％ 的人为（开发者）导致错误。*
智能：极佳的编辑器支持。处处皆可自动补全，减少调试时间。
简单：设计的易于使用和学习，阅读文档的时间更短。
简短：使代码重复最小化。通过不同的参数声明实现丰富功能。bug 更少。
健壮：生产可用级别的代码。还有自动生成的交互式文档。
标准化：基于（并完全兼容）API 的相关开放标准：OpenAPI (以前被称为 Swagger) 和 JSON Schema。

fastapi安装

1.安装fastapi: pip install fastapi
2.如果用于生产，那么你还需要一个ASGI服务器，如Uvicorn或Hypercorn: pip install uvicorn

介绍下基础代码结构

在原有的基础上添加baseurl，目的是用代理时更好的区分不同服务

baseurl = '/xxxx'

app = FastAPI(
    title="WB", #swagger 标题
    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

认证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()  # 根据 时间戳生成 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]):
            # 目录不存在创建，makedirs可以创建多级目录
            os.makedirs(os.path.split(filepath)[0])
            # print('os.makedirs')
        with open(filepath, "wb") as f:
            # 2.3 将获取的fileb文件内容，写入到新文件中
            f.write(contents)
      
    message = {
        'msg': 'success'
    }
    event_lib[ID_event] = message
    return {"ID_event": ID_event,'msg':message[msg]}

查询结果

@app.get(baseurl+"/event/Results/",summary='获取事件结果',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}

运行

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