Flask扩展之http客户端开发

Flask 被称为“微框架”。其中的“微”字不代表整个应用只能塞在一个 Python 文件内，也不代表 Flask 功能不强。它表示 Flask 的目标是保持核心简单而又可扩展。它不会替使用者做决定，比如选用何种数据库，使用何种模板引擎等。Flask 通过扩展功能来增加它的功能。扩展之于 Flask，就像第三方库之于 Python，插件之于 Vscode。本文将介绍如何开发一个简单的 Flask 插件：HTTPClient，并将其发布到 Python 官方索引 Pypi（Python Package Index）上。

介绍

Flask 是一个使用 Python 编写的轻量级 Web 应用框架。它基于 Werkzeug WSGI 工具箱和 Jinja2 模板引擎，并使用 BSD 授权。

Flask 被称为“微框架”，因为它使用简单的核心，用扩展增加其他功能。Flask 没有默认使用的数据库、窗体验证工具。然而，Flask 保留了扩增的弹性，可以用 Flask-extension 加入这些功能：ORM、窗体验证工具、文件上传、各种开放式身份验证技术。

HTTP 客户端在 Flask 应用中也是一个比较常见的需求。如果只是请求一两个 HTTP 服务，那么直接使用 requests 包即可搞定，但是如果需要 Flask 应用去访问某些开放或者收费的 HTTP 服务接口时，此时难道还是每次使用 requests 请求完整的 http://ip:port/path ？设置相同的超时时间？

方案比对

上面的需求是有多种实现方案的，暴力点的就是多次调用，其次是封装成 HTTP 客户端，最优的是封装成 Flask 扩展。

多次调用

该方案主要是参考 requests最佳实践，将 requests 库用好即可实现该功能。

 1import request
 2from requests.adapters import HTTPAdapter
 3import json
 4
 5s = requests.Session()
 6
 7# 设置请求的 header
 8session.headers.update(
 9    {
10        "Content-Type": "application/json",
11        "Referer": "https://httpbin.org/"
12    }
13)
14# 设置请求失败重试次数
15adapter = HTTPAdapter(max_retries=3)
16session.mount('https://', adapter)
17session.mount('http://', adapter)
18# GET,POST请求设置超时时间
19host = 'http://ip:port'
20s.get(url + '/cookies/set/sessioncookie/123456789', timeout=1)
21s.post(url + '/cookies/1',data=json.dumps({'a'：'a'}), timeout=1)

该种方案的特点就是简单粗暴，面向过程编程。

HTTP 客户端

该方案是上面方案的升级版，对上面不同的请求采用面向对象的思想进行封装。

 1import requests
 2
 3import logging
 4logger = getLogger("service")
 5logger.setLevel("INFO")
 6logger.handlers.append(logging.StreamHandler())
 7
 8class HTTPClient(object):
 9    def __init__(self, base_url=None, timeout=None, **kwargs):
10        self.base_url = base_url
11        self.timeout = timeout
12        self.session = requests.Session()
13
14        # request请求重试
15        if self.kwargs.get('retry'):
16            request_retry = requests.adapters.HTTPAdapaters(
17                max_retries=self.kwargs['retry'])
18            self.session.mount('https://', request_retry)
19            self.session.mount('http://', request_retry)
20
21
22    def _request_wrapper(self, method, api, **kwargs):
23        url = self.base_url + api
24        logger.info(
25            f"sending {method} request to {self.url + api} ...  kwargs is {repr(kwargs)}")
26
27        res = self.session.request(method, self.url + api, **kwargs)
28        if res.status_code != 200:
29            raise Exception(f"Http status code is not 200, status code {res.status_code}, "
30                            f"response is {res.content}")
31        # 返回有可能不是json格式
32        if 'text/html' in res.headers['Content-Type']:
33            logger.info(f"sending {method} request to {self.url + api} over ... response is "
34                                 f"{repr(res.content)}")
35            return res.text
36        else:
37            logger.info(f"sending {method} request to {self.url + api} over ... response is "
38                                 f"{repr(res.json())}")
39            return res.json() or dict()
40
41        return self.session.request(method, url, **kwargs)
42
43    def get(self, api, **kwargs):
44        return self._request_wrapper('GET', api, **kwargs)
45
46    def options(self, api, **kwargs):
47        return self._request_wrapper('OPTIONS', api, **kwargs)
48
49    def head(self, api, **kwargs):
50        return self._request_wrapper('HEAD', api, **kwargs)
51
52    def post(self, api, **kwargs):
53        return self._request_wrapper('POST', api, **kwargs)
54
55    def put(self, api, **kwargs):
56        return self._request_wrapper('PUT', api, **kwargs)
57
58    def patch(self, api, **kwargs):
59        return self._request_wrapper('PATCH', api, **kwargs)
60
61    def delete(self, api, **kwargs):
62        return self._request_wrapper('DELETE', api, **kwargs)
63
64    def __del__(self):
65        try:
66            if hasattr(self, "session"):
67                self.session.close()
68        except Exception as e:
69            logger.exception(e)

该方案将需求抽象成一个 HTTPClient 对象，有如下优点：

对象初始化时增加了服务地址base_url，超时timeout，请求重试retry等参数统一设置
使用_request_wrapper函数来统一处理各类请求和处理响应结果
引入日志，方便后续定位解决问题
对象销毁时会关闭打开的请求 session

Flask-HTTPClient

Flask扩展

HTTPClient 类基本能解决大部分问题，但是为什么要做成 Flask 扩展？其实这和 Flask 开发思想：应用工厂和集成扩展有关系。

我们经常在 Flask 的官方帮助文档中看到如下的实例代码。

 1from flask import Flask
 2from flask_sqlalchemy import SQLAlchemy
 3from config import config
 4
 5# 扩展
 6db = SQLAlchemy()
 7
 8# 应用工厂
 9def create_app(config_name):
10    app = Flask(__name__)
11    app.config.from_object(config[config_name])
12
13    # 初始化 db 配置
14    db.init_app(app)
15
16    return app

其中 create_app 函数叫应用工厂函数，是专门用来创建应用的，当然我们可以创建多个应用。db 是关系型数据库ORM的扩展，之所以将其定义在应用工厂函数之外，是为了希望这个扩展实例能够被多个应用使用。换而言之，不同的应用可以挑选不同的扩展组成特定功能的应用。这个就好比 vscode 只是一款编辑器，配上不同编程语言的扩展就可以变成对应编程语言的 IDE。

扩展实现

其实将 HTTPClient 升级为 Flask-HTTPClient 很简单，只需要实现 init_app 函数即可。

 1import requests
 2
 3class HTTPError(Exception):
 4    ...
 5
 6class HTTPClient(object):
 7    def __init__(self, app=None, base_url=None, timeout=None, config_prefix='HTTP_CLIENT',  **kwargs):
 8        self.base_url = base_url
 9        self.timeout = timeout
10        self.config_prefix = config_prefix
11        self.other = kwargs
12
13        if app is not None:
14            self.init_app(app)
15
16    def init_app(self, app):
17        if self.base_url is None:
18            self.base_url = app.config[f'{self.config_prefix}_URL']
19        if self.timeout is None:
20            self.timeout = app.config.get(f'{self.config_prefix}_TIMEOUT', 1)
21        self.session = requests.Session()
22
23        # request请求重试
24        if self.other.get('retry'):
25            request_retry = requests.adapters.HTTPAdapaters(
26                max_retries=self.other['retry'])
27            self.session.mount('https://', request_retry)
28            self.session.mount('http://', request_retry)
29        self.app = app
30
31    def _request_wrapper(self, method, api, **kwargs):
32        url = self.base_url + api
33        self.app.logger.info(
34            f"sending {method} request to {self.url + api} ...  kwargs is {repr(kwargs)}")
35
36        res = self.session.request(method, self.url + api, **kwargs)
37        if res.status_code != 200:
38            raise HTTPError(f"Http status code is not 200, status code {res.status_code}, "
39                            f"response is {res.content}")
40        # 返回有可能不是json格式
41        if 'text/html' in res.headers['Content-Type']:
42            self.app.logger.info(f"sending {method} request to {self.url + api} over ... response is "
43                                 f"{repr(res.content)}")
44            return res.text
45        else:
46            self.app.logger.info(f"sending {method} request to {self.url + api} over ... response is "
47                                 f"{repr(res.json())}")
48            return res.json() or dict()
49
50        return self.session.request(method, url, **kwargs)
51
52    def get(self, api, **kwargs):
53        return self._request_wrapper('GET', api, **kwargs)
54
55    """
56    其它方法和 get 类似
57    """

在上述实现中，主要实现了 init_app 函数，它会将 HTTPClient 实例“加载”到 app 中。此外为了能够共用应用的日志管理，将 app 赋值给 self.app。这样通过 self.app.logger 就可以在扩展中使用应用的日志管理。

发布到Pypi

构建 Flask 扩展 Flask-HTTPClient 的另一个优势就是可以将其发布到 Pypi 上，给广大的 Flask 应用添加候选扩展，避免使用者再重复造轮子。

要想将该扩展发布到 Python 官方索引 Pypi 上，需要组织项目目录如下所示（最终版本见 Github 仓库]）：

1<my_project>/                   # 项目根目录
2|-- <my_package>                # package
3|   |-- __init__.py
4|   |-- <files> ....            # 代码模块
5|-- README.md                   # 帮助文档
6|-- LICENSE                     # 开源协议
7|-- setup.cfg
8|-- setup.py                    # 打包分发配置

当然，如果代码模块就一个文件，可以不采用包模式。

打包发布

打包需要依赖 setuptools 和 wheel 库。而发布需要依赖 twine 这个库。这里我采用 Pipfile 来管理项目的库依赖，使用 Makefile 来管理常用命令。

 1# 安装 pipenv 库，并安装该项目所需依赖
 2make deploy
 3
 4# 打包
 5make build
 6
 7# 发布
 8make publish
 9
10# 清理环境
11make clean

当然在发布前需要到官方网站 Pypi 上注册一个账号，在执行发布命令时要输入用户名和密码。最终就能在官网上看到自己发布的 Flask扩展 HTTPClient了。广大的 Flask 用户可以通过以下命令来安装该扩展：

1pip install Flask-HTTPClient

也可以看看