Flyer：基于FastAPI的轻量级API开发框架

概述

我们经常会听到“不会开发的运维不是个好运维”，确实在运维工作当中，娴熟的开发经验能让运维工作如鱼得水，事半功倍！高级的运维除了需要精通 Shell 脚本编程之外，最好还需要掌握一门高级语言，比如 Python、Golang、PHP、Lua 等。高级语言除了可以解决一些更复杂的运维场景，还可以帮助我们更好的理解业务，做好业务运维，毕竟知己知彼才能百战不殆。

经常看到很多运维同学写接口通常是基于 Flask，只用一个脚本一撸到底，完全没有框架的概念，总体给人的感觉比较粗糙简陋。本文分享一个我个人基于 FastAPI 框架设计的轻量级 API 开发框架，能够应付绝大部分的业务运维的后台 API 场景，这个框架我取名为 Flyer，意味着轻量且快速。

框架特性

• 继承 FastAPI 所有特性，包括高性能、自带 Swagger/ReDoc 文档、易开发等，详见：(FastAPI (tiangolo.com))
• 拥抱云原生，所有参数都可以通过七彩石/环境变量来自定义，代码中也可以非常方便的获取指定环境变量的值
• 鹅厂内部版本集成了七彩石、智研日志汇、监控宝等各种研效库（外部版本已剥离）
• MySQL/Redis/Kafka 等开源组件已默认对接，可以开箱即用
• 支持记录全局请求/被请求流水日志及耗时等监控特性
• 可以快速开启简单的 BasicAuth 鉴权等
• 更多细节这里省略 1 万字...

性能测试

虽然这个框架主要应对的是运维场景，性能并不是主要矛盾，但想到很多同学还是会比较感兴趣，这里简单测试了一下单核空载性能，命令行如下：

# 基于已有镜像快速启动
docker run -d \
    --net=host \
    --name flyer_bench \
    --cpus 1 \
    -e flyer_port=8888 \
    -e flyer_workers=1 \
    -e flyer_log_level=error \
    -e flyer_debug=False \
    -e flyer_reload=False \
    -e flyer_preload=True \
    -e flyer_threads=1 jagerzhang/flyer:v1.0

# 使用 wrk 对监控检查接口发起压测
./wrk -c 800 -t 10 http://127.0.0.1:8888/health_check

结果如下：

从结果可以看到，单 CPU 空载可以跑到 8000+QPS，这个性能应该能应付绝大部分中大型运维场景了。当然，这里测试的接口是开启了 async 异步协程的性能极限模式，改为 sync 的话，性能也能跑到 4k~5k 左右。

快速上手

安装 Docker

略，这个应该都会。

构建镜像

git clone https://github.com/jagerzhang/flyer.git
cd flyer
docker build -t "flyer:test" .

启动服务

docker run \
  --net=host \
  --name=flyer \
  flyer:test ./start-dev.sh

验证服务

浏览器打开如下地址可以看到效果（<IP>换成部署服务的 IP 地址），如下图所示：

• ReDoc: http://<IP>:8080/flyer/v1/redoc
• SwaggerUI: http://<IP>:8080/flyer/v1/dcos

正式开发

容器模式

可以基于上文制作的 flyer:test 容器镜像来快速部署开箱即用的开发环境：

# 克隆代码
cd /data/
git clone https://github.com/jagerzhang/flyer.git

# 启动容器，将本地代码挂进去
docker run \
--net=host \
--name=flyer_dev \
-v /data/flyer:/flyer \
flyer:test \
./start-dev.sh

普通模式

安装依赖

# 安装 Python3 和基础组件
yum install -y python3 python3-devel python3-setuptools make snappy-devel gcc-c++

# 安装 Flyer 依赖的 Python 插件
pip3 install --no-cache-dir -r requirements.txt 

启动服务

git clone https://github.com/jagerzhang/flyer.git
cd flyer
./start-dev.sh

成功启动后，浏览器访问以下地址即可查看效果，修改任何 Python 代码，保存后都会自动重新加载，非常方便：

• ReDoc: http://<IP>:8080/flyer/v1/redoc
• SwaggerUI: http://<IP>:8080/flyer/v1/dcos

目录结构

以下是框架的代码结构说明，开发时只需要复制一份 api/demo 文件夹进行逻辑代码编写：

.
├── api                            # API 汇总目录， 里面按文件夹存放独立的 API 服务
│   ├── base                       # 框架内置 API，主要包括监控检查、swagger 文档、ReDoc 文档
│   ├── demo                       # 内置的 API 样例，用于 API 开发参考 
│   │   ├── __init__.py            # 注册路由
│   │   ├── models                 # API 核心逻辑
│   │   ├── routers                # 定义各接口路由
│   │   ├── schemas                # 定义各接口协议
│   │   └── settings.py            # 当前 API 的自定义配置
│   ├── __init__.py                # 总 API 加载入口
│   └── settings.py                # 全局配置脚本
├── build_base.sh                  # 基础镜像构建脚本
├── build.yaml                     # PreCI 本地代码检查配置
├── docker                         # Docker 启动脚本
├── Dockerfile                     # 服务镜像构建配置
├── Dockerfile_base                # 基础镜像构建配置
├── logs                           # API 日志目录
├── main.py                        # 开发环境启动脚本(被 start-dev.sh 调用)
├── README.md                  
├── requirements.txt               # 框架依赖
├── run.sh                         # 容器服务启动入口脚本
├── start-dev.sh                   # 开发环境启动入口脚本
├── static                         # 静态文件目录
├── tests                          # 单元测试脚本目录
│   ├── __init__.py              
│   ├── start-test.sh              # 手工发起一键测试(非 Pytest 方式)
│   ├── test_base.py               # 内置接口的测试脚本
│   ├── test_demo.py               # demo 接口的测试脚本
│   ├── test_exception_validate.py # 参数验证异常的测试脚本
│   └── test_health.py             # 健康检查接口的测试脚本
└── utils                          # 框架公共方法目录
    ├── common.py                  # 通用方法函数
    ├── data.py                    # Redis、MySQL 连接池
    ├── http_requests.py           # 对外 HTTP 请求公共函数，可以记录日志和异常重试
    ├── ierror.py                  # 接口返回码的定义脚本
    ├── __init__.py
    ├── logger.py                  # 日志初始化
    ├── middleware.py              # 框架中间件逻辑，用于埋点、记录耗时、日志等

开发步骤

首先复制根目录下的 api/demo 文件夹到一个新的文件夹，比如 myapp（下文的介绍均按这个名字），然后按照以下步骤来开发即可：

定义接口协议

打开 myapp.schemas 文件夹下的 demo.py 文件，参考已有内容去定义接口字段，包括字段名称、字段属性及描述等：

# -*- coding: utf-8 -*-
"""
参数验证模块
"""
from pydantic import BaseModel, Field
from api.settings import ierror


class DemoRequest(BaseModel):
    """ Demo 演示：请求参数.
    """
    msgContent: str = Field(example="Flyer", title="Flyer 演示项目")


class DemoResponse(BaseModel):
    """ Demo 演示：响应参数.
    """
    retInfo: str = Field(default="Hello Flyer!",
                         example="Hello Flyer!",
                         title="Flyer 演示项目返回信息")
    retCode: int = Field(
        default=ierror.IS_SUCCESS,
        example=ierror.IS_SUCCESS,  # NOQA
        title="Flyer 演示项目返回码")

开发接口逻辑

打开 myapp.models 文件夹下的 demo.py 文件，这里写接口逻辑代码，这个按实际需求开发即可：

# -*- coding: utf-8 -*-
""" 功能逻辑模块
"""
 
class DemoClass():
    """ 示例方法
    """
    def __init__(self):
        """Codding Here"""
        pass
 
    def demo_func(self, user, msg):
        result = {"msgContent": f"{user}，你好！已成功收到你的指令：{msg}"}
        return result

定义接口路由

打开 myapp.routes 文件夹下的 demo.py 文件，按需修改：

from fastapi import APIRouter, Request
from api.demo.schemas.demo import DemoRequest, DemoResponse
from utils.middleware import RouteMiddleWare
from api.settings import ierror

router = APIRouter(route_class=RouteMiddleWare)


@router.post("/demo", response_model=DemoResponse, summary="Demo")
async def demo(params: DemoRequest, request: Request):
    """
    Demo 演示接口
    ---
    - 功能说明: 用于演示 Flyer 开发框架，传入一个名字，返回 "Hello <名字>!"
    - 附加说明 1: 详细的参数说明可以查看<a href="/flyer/v1/redoc#tag/Demo" \
target="_blank">接口文档</a>；
    - 附加说明 2: 这个位置可以加入更多说明列表。
    """
    result = {
        "retCode": ierror.IS_SUCCESS,
        "retInfo": f"Hello {params.msgContent}!"
    }
    return 

然后，编辑 myapp.routes.__init__.py，注册路由到 APIRouter：

from fastapi import APIRouter
from api.demo.routers import demo

demo_api = APIRouter()
demo_api.include_router(demo.router, tags=["Demo 接口"])

最后，编辑 api.__init__.py，将新路由注册到 fastapi：

# ... 上面略
from api.myapp.routers import demo_api

# 内容略
def create_app():
    """加载应用入口
    """
    # 内容略

    # 加入新接口路由注册，这里可以控制是否鉴权，可以参考下文内容
    # 是否开启鉴权
    if int(config.env_list.get("flyer_auth_enable", 0)) == 1:
        app.include_router(demo_api,
                           prefix=f"{config.base_url}/{config.version}",
                           dependencies=[Depends(authorize)])
    else:
        app.include_router(demo_api,
                           prefix=f"{config.base_url}/{config.version}")

    create_service(app)
    register_exception(app)
    return app

定义接口鉴权

如果你对于接口安全有要求，这个框架也支持快速开启 BasicAuth 接口鉴权，需要在启动框架之前，如下设置环境变量：

 打开鉴权
export flyer_auth_enable=1
# 定义鉴权帐号
export flyer_auth_user=user
# 定义鉴权密码
export flyer_auth_pass=pass

然后参考 utils.authorize.py 文件注释，并对比下 api.__init__.py 36-43 行代码逻辑，给新的路由加上权限限制即可。

接口文档

Flyer 基于 FastAPI 框架，所以自带了 reDoc 和 SwaggerUI，完全实现代码即文档的开发方式，非常方便。

• ReDoc: http://<IP>:8080/flyer/v1/redoc
• SwaggerUI: http://<IP>:8080/flyer/v1/dcos

环境变量

Flyer 支持通过环境变量来修改各种配置。

基础变量

• flyer_host: 接口绑定 IP，默认为 0.0.0.0
• flyer_port：接口绑定端口，默认为 8080
• flyer_base_url：服务地址前缀，默认为 /flyer
• flyer_version：服务版本，当前为 v1
• flyer_reload：接口热加载，用于开发环境，默认为 True
• flyer_workers：工作进程数量，默认为 1
• flyer_threads：工作线程数量，默认为 5
• flyer_worker_connections：最大客户端并发数量，默认为 1000
• flyer_enable_max_requests：打开自动重启机制，即请求一定数量后进程自动重启，可以缓解内存泄漏问题
• flyer_max_requests：重新启动之前，工作将处理的最大请求数。默认值为 0
• flyer_max_requests_jitter：要添加到 max_requests 的最大抖动。抖动将导致每个工作的重启被随机化，这是为了避免所有工作被重启。
• flyer_timeout：超过这么多秒后工作将被杀掉，并重新启动。默认为 60 秒
• flyer_graceful_timeout：优雅退出时间，默认为 10，收到重启信号后，将等待指定时长才（或强制）退出
• flyer_keepalive：在 keep-alive 连接上等待请求的秒数，默认为 5
• flyer_log_level：定义日志级别，debug/info/warn/error，默认为 info
• flyer_access_log：是否记录请求日志，True/False，默认为 True
• flyer_access_logfile：定义请求日志文件的位置，默认为-，即输出到容器标准输出

按需变量

需要用到 Redis 请添加如下配置：

• flyer_redis_host: Redis 服务 IP，默认 localhost
• flyer_redis_port: Redis 服务端口，默认 6379
• flyer_redis_pass: Redis 服务密码，默认为空
• flyer_redis_db: Redis 实例 DB，默认为 1

需要用到 MySQL 请添加如下配置：

• flyer_db_host: MySQL 服务地址
• flyer_db_port: MySQL 服务端口
• flyer_db_user: MySQL 用户名
• flyer_db_pass: MySQL 密码
• flyer_db_name: MySQL 数据库名称

需要用到 kafka 请添加如下配置：

• flyer_kafka_topic: 指定 kafka 消息队列的 Topic
• flyer_kafka_servers: 指定 kafka 消息队列的 服务地址，格式为 x.x.x.x:9092,y.y.y.y:9092

单元测试

Flyer 的单元测试用例位于根目录下的 tests 目录，支持 pytest ，完成接口开发后，建议参考 tests/test_demo.py 快速补齐测试用例，提高单侧覆盖率。

手工测试

测试单个用例

cd tests
./start-test.sh test_demo.py

测试所有用例

cd tests
./start-test.sh

PyTest 测试

# 安装 pytest 和覆盖率统计插件
pip install pytest coverage
cd tests

# 测试单个用例
pytest test_demo.py

# 测试所有用例
pytest

# 测试所有用例，同时展示测试用例的覆盖率统计
pytest --cov=. tests/

写在最后

因文章篇幅有限，很多细节都没能一一介绍，这部分可以在使用过程中通过阅读代码来慢慢熟悉，如有疑问也可以留言咨询。Flyer 是我纯个人兴趣、为满因文章篇幅有限，框架的很多细节都没能一一介绍，这部分可以在使用过程中通过阅读代码来慢慢熟悉，如有疑问也可以直接企微咨询。Flyer 是我个人兴趣、为满足日常系统运维工作而设计的 Python 开发框架，因为我本身并非专业开发，所以这个框架肯定有很多不专业的地方，这里整理分享出来也是想给有相同需求的系统运维同学提供一些便利。如果在试用过程中发现 Bug 或有更好的优化建议可以直接提交 issue 或 PR，参与开源共建。