干貨：使用Fastapi開發自己的Mock server(附原始碼)

最近幾天抽空看了下Fastapi的檔案，順便使用它開發了個HTTP(s)協議的Mock server，以后有需要用到擋板的地方，就可以直接使用啦，

專案一級目錄：

logs用來存放日志檔案，日切

main：啟動入口程式，直接運行

requirements.txt：依賴包

common中存放日志模塊，核心模塊等公用部分，

apis用來存放各個專案的擋板服務，各目錄以專案為單位分開；每個專案目錄下固定有三個檔案夾：

api_operation：存放接收post、get等請求并回傳Mock回應的代碼，

config：存放Mock的匹配規則

responce_file：存放Mock回應的檔案，json、txt.

注意：config下的組態檔名與responce_file要一一對應，

api_operation的開發

import timefrom fastapi import APIRouter, Path, Queryfrom starlette.responses import FileResponsefrom starlette.requests import Requestfrom common.mock_responce import MockResponcefrom common.log_config import getlogerrouter = APIRouter()logger = getloger(name)@router.post("/v1/{url_path}")async def json_res(request: Request, url_path=Path(…, title=“The ID of the item to get”)): print(request.url) # query_args = request.query_params # logger.info(‘query_args:’, query_args) # logger.info(query_args.get(‘entInfo’)) # form_args = await request.form() # print(‘form_args:’, form_args) # print(form_args.get(‘entInfo’)) # byte_body = await request.body() # try: # import json # b_body = json.loads(byte_body.decode(‘UTF-8’)) # print(‘byte_body:’, b_body) # print(type(b_body)) # except: # b_body = byte_body.decode(‘UTF-8’) # print(‘byte_body:’, b_body) # print(type(b_body)) json_body = await request.json() return FileResponse(MockResponce(‘test_api’, ‘A001’, json_body).responce_filter())@router.post("/v1/txt")async def txt_res(request: Request): form_args = await request.form() return FileResponse(MockResponce(‘test_api’, ‘A002’, form_args).responce_filter())
通過request引數可以獲取請求傳進來的所有header、引數等；

引數通常有三種方法傳進來：

1.url傳參，query_args = request.query_params來獲取

2.form-da傳參，form_args = await request.form()來獲取

3.body傳參，如果穿的是json或json格式的字串，用json_body = await request.json()來獲取；如果穿的是二進制或類json字串，可以使用byte_body = await request.body()來獲取，

然后將取得的值傳給核心模塊即可：

MockResponce(‘test_api’, ‘A001’, json_body).responce_filter()

test_api：擋板專案名，即apis下一級目錄的名稱，

A001：規則檔案名

json_body：從request所取得的引數

技術細節:

1,2兩種方式接收到的引數雖然不是字典，但是都可以使用跟字典一樣的方式取得引數，例如，url傳參如下：

http://127.0.0.1:8000/vba1/vvv/23123?entInfo=平安銀行股份有限公司&appid=weq12311

可以使用query_args[‘entInfo’]或者query_args.get(‘entInfo’)來獲取entInfo的值，

這樣核心模塊就可以將取得的entInfo與規則檔案進行匹配，并根據匹配的結果獲得相應檔案回傳，

3這種方式接收的引數，大部分為json格式，可以直接轉化為字典，但是在復雜嵌套的json中想要取得entInfo的值，就更復雜，Mock server核心部分已經做了封裝，會自動處理，少部分body傳進來的可能是二進制，需根據實際情況再處理，

服務路由
以上開發完成后，需要將服務添加路由至fastap的app，類似于Flask的Blueprint,具體位置如下圖

在這里還可以設定請求路徑共同的部分prefix等，

from fastapi import FastAPIfrom apis.test_api.api_operation import
study1,study2from apis.CIF_api.api_operation import
bindcard,evalPhonefrom apis.vba_api.api_operation import vba1,vba2def
create_app(): app = FastAPI(title=‘FastApi Mock Server’,
description=‘這是使用fastapi開發的擋板服務程式，旨在使用簡單的操作實作快速可用的擋板服務’,
version=0.1) app.include_router( study1.router,
prefix="/items", tags=[“items”], dependencies=[],
responses={404: {“description”: “Not found”}}, )
app.include_router( study2.router, prefix="/test",
tags=[“test”], dependencies=[], responses={404:
{“description”: “Not found”}}, ) app.include_router(
bindcard.router, prefix="/bindcard", tags=[“bindcard”],
dependencies=[], responses={404: {“description”: “Not found”}},
) app.include_router( evalPhone.router,
prefix="/evalPhone", tags=[“evalPhone”],
dependencies=[], responses={404: {“description”: “Not found”}},
) app.include_router( vba1.router, prefix="/vba1",
tags=[“vba1”], dependencies=[], responses={404:
{“description”: “Not found”}}, ) app.include_router(
vba2.router, prefix="/vba2", tags=[“vba2”],
dependencies=[], responses={404: {“description”: “Not found”}},
) return app

config下規則檔案的定義
#入參為空，value:None#vba1即該檔案的名字，必須保持一致，純數字要加引號，不然會被yaml當做intvba1: - res_file: 平安銀行股份有限公司.json enable: Y index: 1 delay: 0 condition: - {field: entInfo,rule: is,value: 平安銀行股份有限公司} - {field: appid,rule: is,value: weq12313331} - res_file: 查詢無記錄.json enable: Y index: 2 delay: 0 condition: - {field: entInfo,rule: is,value: 查詢無記錄} - res_file: 入參為空.json enable: Y index: 3 delay: 0 condition: - {field: entInfo,rule: is,value: None} - res_file: 文本回應.txt enable: Y status: 1 index: 7 delay: 0 condition: - {field: entInfo,rule: regex,value: ‘.+’}
res_file:responce_file對應目錄下存放Mock回應的檔案名，必須帶后綴，

enable：是否啟用該規則，Y,y啟用，N，n不啟用

index：規則序號

delay：回應延時，為空默認為0，例如delay: 5，如果匹配到這條規則，則延時5秒再回傳回應，

condition：匹配該規則的條件，如果多個條件為且的關系，所有規則并列放在一起，必須滿足全部；如果多條規則為或的關系，則所有規則分開寫，命中任意一條即回傳，

field：需要用來匹配的關鍵字，如上則表示需要從請求的引數中拿到entInfo的值

rule：匹配規則，is-完全一樣；regex-正則匹配

value：規則的期望值，field中取得的值與value中的值匹配，value: None表示傳入的引數為慷訓未傳入該引數，

規則匹配順序，按照規則檔案中寫的順序，從上到下，命中即回傳，

技巧：

在規則檔案末尾一般放一個全匹配的正則規則兜底，保證任何請求都有回應：

  - res_file: 文本回應.txt    enable: Y    status: 1    index: 7    delay: 0    condition:      - {field: entInfo,rule: regex,value: '.+'}

服務啟動

以上都開發好后，使用時，在main.py同級目錄的命令列運行：

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

啟用HTTPS:

uvicorn main:app --host 0.0.0.0 --port 8000 --reload --ssl-keyfile server.key --ssl-certfile server.crt

如果要啟用http，則加入server.crt與server.key啟動服務，并將ca.crt追加到客戶端linux服務器的授信串列：

cat ca.crt>> /etc/pki/tls/certs/ca-bundle.crt

或者直接運行main.py啟動服務

import uvicornfrom fastapi import Depends, FastAPI, Header, HTTPExceptionfrom apis import create_appfrom common.log_config import getlogerapp = create_app()logger = getloger(__name__)@app.on_event("shutdown")def shutdown_event():    print('shut down')@app.on_event("startup")async def startup_event2():    print('startup')if __name__ == "__main__":    uvicorn.run('main:app', host="0.0.0.0", port=8000, reload=True)    # uvicorn.run('main:app', host="0.0.0.0", port=8000, reload=True, ssl_keyfile='server.key', ssl_certfile='server.crt')

Mock server使用

使用時，只需要修改規則yaml檔案和responce_file回應檔案即可，

例如，對于上面的規則，body給不同的傳參效果：

此外，由于fastapi的特點，還可以使用自帶的swagger服務查看所有擋板服務：

代碼已上傳中github:http://github.com/superxuu/fastapi_mock

今天分享就到這里，喜歡的點個關注~~

最后： 給大家推薦一個 q 群：902061117 里面有許多資料共享！資料都是面試時面試官必問的知識點，也包括了很多測驗行業常見知識，其中包括了有基礎知識、Linux必備、Shell、互聯網程式原理、Mysql資料庫、抓包工具專題、介面測驗工具、測驗進階-Python編程、Web自動化測驗、APP自動化測驗、介面自動化測驗、測驗高級持續集成、測驗架構開發測驗框架、性能測驗、安全測驗等，

如果對你有一點點幫助，各位的「點贊」就是小編創作的最大動力，我們下篇文章見！

好文推薦

2021軟體測驗工程師面試題匯總（內含答案）-看完BATJ面試官對你豎起大拇指！

什么樣的人適合從事軟體測驗作業？

軟體測驗和軟體開發哪個發展更好

那個準點下班的人，比我先升職了…