Flask扩展之http客户端开发
Flask 被称为“微框架”。其中的“微”字不代表整个应用只能塞在一个 Python 文件内，也不代表 Flask 功能不强。它表示 Flask 的目标是保持核心简单而又可扩展。它不会替使用者做决定，比如选用何种数据库，使用何种模板引擎等。Flask 通过扩展功能来增加它的功能。扩展之于 Flask，就像第三方库之于 Python，插件之于 Vscode。本文将介绍如何开发一个简单的 Flask 插件：HTTPClient，并将其发布到 Python 官方索引 Pypi（Python Package Index） 上。
介绍
Flask[1] 是一个使用 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最佳实践[2]，将 requests 库用好即可实现该功能。
import request
from requests.adapters import HTTPAdapter
import json
s = requests.Session()
# 设置请求的 header
session.headers.update(
{
"Content-Type": "application/json",
"Referer": "https://httpbin.org/"
}
)
# 设置请求失败重试次数
adapter = HTTPAdapter(max_retries=3)
session.mount('https://', adapter)
session.mount('http://', adapter)
# GET,POST请求设置超时时间
host = 'http://ip:port'
s.get(url + '/cookies/set/sessioncookie/123456789', timeout=1)
s.post(url + '/cookies/1',data=json.dumps({'a'：'a'}), timeout=1)
该种方案的特点就是简单粗暴，面向过程编程。
HTTP 客户端
该方案是上面方案的升级版，对上面不同的请求采用面向对象的思想进行封装。
import requests
import logging
logger = getLogger("service")
logger.setLevel("INFO")
logger.handlers.append(logging.StreamHandler())
class HTTPClient(object):
def __init__(self, base_url=None, timeout=None, **kwargs):
self.base_url = base_url
self.timeout = timeout
self.session = requests.Session()
# request请求重试
if self.kwargs.get('retry'):
request_retry = requests.adapters.HTTPAdapaters(
max_retries=self.kwargs['retry'])
self.session.mount('https://', request_retry)
self.session.mount('http://', request_retry)
def _request_wrapper(self, method, api, **kwargs):
url = self.base_url + api
logger.info(
f"sending {method} request to {self.url + api} ... kwargs is {repr(kwargs)}")
res = self.session.request(method, self.url + api, **kwargs)
if res.status_code != 200:
raise Exception(f"Http status code is not 200, status code {res.status_code}, "
f"response is {res.content}")
# 返回有可能不是json格式
if 'text/html' in res.headers['Content-Type']:
logger.info(f"sending {method} request to {self.url + api} over ... response is "
f"{repr(res.content)}")
return res.text
else:
logger.info(f"sending {method} request to {self.url + api} over ... response is "
f"{repr(res.json())}")
return res.json() or dict()
return self.session.request(method, url, **kwargs)
def get(self, api, **kwargs):
return self._request_wrapper('GET', api, **kwargs)
def options(self, api, **kwargs):
return self._request_wrapper('OPTIONS', api, **kwargs)
def head(self, api, **kwargs):
return self._request_wrapper('HEAD', api, **kwargs)
def post(self, api, **kwargs):
return self._request_wrapper('POST', api, **kwargs)
def put(self, api, **kwargs):
return self._request_wrapper('PUT', api, **kwargs)
def patch(self, api, **kwargs):
return self._request_wrapper('PATCH', api, **kwargs)
def delete(self, api, **kwargs):
return self._request_wrapper('DELETE', api, **kwargs)
def __del__(self):
try:
if hasattr(self, "session"):
self.session.close()
except Exception as e:
logger.exception(e)
该方案将需求抽象成一个 HTTPClient 对象，有如下优点：
Flask-HTTPClient
Flask扩展
HTTPClient 类基本能解决大部分问题，但是为什么要做成 Flask 扩展？其实这和 Flask 开发思想：应用工厂和集成扩展有关系。
我们经常在 Flask 的官方帮助文档中看到如下的实例代码。
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from config import config
# 扩展
db = SQLAlchemy()
# 应用工厂
def create_app(config_name):
app = Flask(__name__)
app.config.from_object(config[config_name])
# 初始化 db 配置
db.init_app(app)
return app
其中 create_app 函数叫应用工厂函数，是专门用来创建应用的，当然我们可以创建多个应用。db 是关系型数据库ORM的扩展，之所以将其定义在应用工厂函数之外，是为了希望这个扩展实例能够被多个应用使用。换而言之，不同的应用可以挑选不同的扩展组成特定功能的应用。这个就好比 vscode 只是一款编辑器，配上不同编程语言的扩展就可以变成对应编程语言的 IDE。
扩展实现
其实将 HTTPClient 升级为 Flask-HTTPClient 很简单，只需要实现 init_app 函数即可。
import requests
class HTTPError(Exception):
...
class HTTPClient(object):
def __init__(self, app=None, base_url=None, timeout=None, config_prefix='HTTP_CLIENT', **kwargs):
self.base_url = base_url
self.timeout = timeout
self.config_prefix = config_prefix
self.other = kwargs
if app is not None:
self.init_app(app)
def init_app(self, app):
if self.base_url is None:
self.base_url = app.config[f'{self.config_prefix}_URL']
if self.timeout is None:
self.timeout = app.config.get(f'{self.config_prefix}_TIMEOUT', 1)
self.session = requests.Session()
# request请求重试
if self.other.get('retry'):
request_retry = requests.adapters.HTTPAdapaters(
max_retries=self.other['retry'])
self.session.mount('https://', request_retry)
self.session.mount('http://', request_retry)
self.app = app
def _request_wrapper(self, method, api, **kwargs):
url = self.base_url + api
self.app.logger.info(
f"sending {method} request to {self.url + api} ... kwargs is {repr(kwargs)}")
res = self.session.request(method, self.url + api, **kwargs)
if res.status_code != 200:
raise HTTPError(f"Http status code is not 200, status code {res.status_code}, "
f"response is {res.content}")
# 返回有可能不是json格式
if 'text/html' in res.headers['Content-Type']:
self.app.logger.info(f"sending {method} request to {self.url + api} over ... response is "
f"{repr(res.content)}")
return res.text
else:
self.app.logger.info(f"sending {method} request to {self.url + api} over ... response is "
f"{repr(res.json())}")
return res.json() or dict()
return self.session.request(method, url, **kwargs)
def get(self, api, **kwargs):
return self._request_wrapper('GET', api, **kwargs)
"""
其它方法和 get 类似
"""
在上述实现中，主要实现了 init_app 函数，它会将 HTTPClient 实例“加载”到 app 中。此外为了能够共用应用的日志管理，将 app 赋值给 self.app。这样通过 self.app.logger 就可以在扩展中使用应用的日志管理。
发布到Pypi
构建 Flask 扩展 Flask-HTTPClient 的另一个优势就是可以将其发布到 Pypi 上，给广大的 Flask 应用添加候选扩展，避免使用者再重复造轮子。
要想将该扩展发布到 Python 官方索引 Pypi 上，需要组织项目目录如下所示（最终版本见 Github 仓库[3]]）：
<my_project>/ # 项目根目录
|-- <my_package> # package
| |-- __init__.py
| |-- <files> .... # 代码模块
|-- README.md # 帮助文档
|-- LICENSE # 开源协议
|-- setup.cfg
|-- setup.py # 打包分发配置
当然，如果代码模块就一个文件，可以不采用包模式。
打包发布
打包需要依赖 setuptools 和 wheel 库。而发布需要依赖 twine 这个库。这里我采用 Pipfile 来管理项目的库依赖， 使用 Makefile 来管理常用命令。
# 安装 pipenv 库，并安装该项目所需依赖
make deploy
# 打包
make build
# 发布
make publish
# 清理环境
make clean
当然在发布前需要到官方网站 Pypi[4] 上注册一个账号，在执行发布命令时要输入用户名和密码。最终就能在官网上看到自己发布的Flask扩展 HTTPClient[5]了。广大的 Flask 用户可以通过以下命令来安装该扩展：
pip install Flask-HTTPClient
参考文献
1.维基百科Flask[6]2.flask扩展官方文档[7]3.Flask-HTTPClient[8]4.requests最佳实践[9]5.怎样将Python项目发布到PyPI[10]6.pypi库Flask-HTTPClient[11]7.Python 库打包分发(setup.py 编写)简易指南[12]
References
[1] Flask: https://zh.wikipedia.org/wiki/Flask[2] requests最佳实践: https://requests.readthedocs.io/zh_CN/latest/user/advanced.html[3] Github 仓库: https://github.com/haojunyu/flask-httpclient[4] Pypi: https://pypi.org/[5] Flask扩展 HTTPClient: https://pypi.org/project/Flask-HTTPClient/[6] 维基百科Flask: https://zh.wikipedia.org/wiki/Flask[7] flask扩展官方文档: https://dormousehole.readthedocs.io/en/latest/extensiondev.html[8] Flask-HTTPClient: https://github.com/haojunyu/flask-httpclient[9] requests最佳实践: https://requests.readthedocs.io/zh_CN/latest/user/advanced.html[10] 怎样将Python项目发布到PyPI: https://zhuanlan.zhihu.com/p/37987613[11] pypi库Flask-HTTPClient: https://pypi.org/project/Flask-HTTPClient/[12] Python 库打包分发(setup.py 编写)简易指南: https://blog.konghy.cn/2018/04/29/setup-dot-py/