Python|fastapi从入门到不放弃(二)

之前写了篇fastapi的简介[Python|介绍一下我的新伙伴fastapi(一)],今天直接进入干货阶段,将最近一段时间对fastapi的探索整理下来,也算是总结回顾.

基础配置

修改标题,描述,版本号
代码

app = FastAPI(
    ttitle="雨鸟飞行fastapi示例",
    description="演示fastapi常用的一些配置",
    version="0.1.0",
)

效果

修改swagger路径

默认/dcos访问swagger可以修改别的,比如/documents
代码

app = FastAPI(docs_url="/documents")

效果
访问/docs返回not found

{
detail: "Not Found"
}

隐藏redoc

fastapi默认自带两种接口显示方式:swagger+redoc,对于我来说swagger已经足够用了,没必要显示redoc直接干掉,也能避免一些安全隐患
代码

app = FastAPI(redoc_url=None)

效果
访问/redoc返回not found

{
detail: "Not Found"
}

### 接口分组
默认所有的接口都是放在default下显示,我们可以通过增加tag的方式,将不同类型的接口分组管理,核心在接口后添加tags=['分组名']
默认效果
代码

@app.get("/items/{item_id}",tags=["items"])
async def read_item(item_id: int, q: Optional[str] = None):
    return {"item_id": item_id, "q": q}
@app.put("/items/{item_id}",tags=["items"])
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}

效果

增加自定义提示和链接

其实就是自定义一些tag,如果没有接口关联,就直接显示在那里了
代码

tags_metadata = [
    {
        "name": "介绍",
        "description": "雨鸟飞行的博客,显示fastapi的基本使用",
        "externalDocs": {
            "description": "博客地址",
            "url": "https://blog.cnrainbird.com",
        },
    }
]

app = FastAPI(openapi_tags=tags_metadata)

效果

中间件获取执行时间

log打印每个请求花费的时间,方便排查问题
代码

@app.middleware("http")
async def log_middle(request: Request, call_next):
    start_time = time.time()
    logger.debug(f"{request.method} {request.url}")
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    logger.info(f"{request.method} {request.url} {process_time}")
    return response

效果

swagger隐藏关键接口

有些敏感接口涉及加密信息等的,不在swagger显示,此处以无用的默认/及favicon.ico为例,关键字是include_in_schema=False
代码

@app.get('/favicon.ico',include_in_schema=False)
@app.get('/',include_in_schema=False)

效果
不知道咋截图,就是在swagger看不到这两个了

暴露健康检查接口

每个应用都应该有个接口,直接查询本服务运行状态,依赖的三方是否正常运行
代码

@app.get('/healthz',tags=["health"])
def healthz_check(request:Request):
    res_dict = {}
    res_dict['headers'] = request.headers
    config = []
    cfg_path = os.path.dirname(os.path.abspath(__file__)) + '/config.json'
    if os.path.isfile(cfg_path):
        config.append(f'''ok {cfg_path} exists''')
    else:
        config.append(f'''failed {cfg_path} exists''')
    res_dict['config'] = config
    res_dict['run']    = {}
    res_dict['system'] = {'hostname':'','ip':''}
    res_dict['env'] = os.environ
    return response_struct(data=res_dict)

效果

接口用途和描述

swagger已经很方便展示接口的入参及类型了,如果再加上一些描述岂不更香!
代码

@app.get('/healthz',tags=["health"])
def healthz_check(request:Request):
    '''
    服务健康检查

    留给运维检查服务状态,此处建议代码检查服务正常运行依赖项
    包括:
    - **headers** 请求信息
    - **config** 配置信息
    - **run** 运行依赖配置
    - **redis**  redis是否正常连接
    - **mysql** mysql连接状态
    - **api** 三方api调用是否正常
    '''

效果

后台执行

对于比较耗时的操作,或者前端只是触发一下并不需要立即返回结果,或者是定时任务只需要触发一下后台慢慢执行等
代码

from fastapi import BackgroundTasks
def do_send_mail(mail_to,mail_body):
    logger.info(mail_to,mail_body)
    time.sleep(3)
    logger.info(mail_to,mail_body)
@app.get('/send_mail',tags=['backgound'])
def send_mail_to_someone(mail_to:str,mail_body:str,background_tasks: BackgroundTasks):
    '''
    异步发送邮件

    比较耗时,后台rv执行
    '''
    background_tasks.add_task(do_send_mail,mail_to,mail_body)
    return response_struct()

效果
可以看到两个mail_to的打印时间中间差了3秒,但前端立即返回了

结尾

在fastapi的使用上,我还是小学生,也会一直继续使用下去,有新的心得会继续发出来共享

fastapi


相关博文

About rainbird

IOS攻城狮
This entry was posted in Python and tagged , , , , , , , , , , , , . Bookmark the permalink.

发表评论