vlambda博客
学习文章列表

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/


推荐阅读

[5.05] 暗喻幻想港中35周年限定版1.03+DLC+金手指+王中王存档 [5.05] 暗喻幻想港中35周年限定版1.03+DLC+金手指+王中王存档
“潜伏”8年，FLASK凡氪能否带来电子雾化市场“不凡一刻”？ “潜伏”8年，FLASK凡氪能否带来电子雾化市场“不凡一刻”？

相关文章