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 库用好即可实现该功能。

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 对象,有如下优点:

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

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 仓库]):

<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 上注册一个账号,在执行发布命令时要输入用户名和密码。最终就能在官网上看到自己发布的Flask扩展 HTTPClient了。广大的 Flask 用户可以通过以下命令来安装该扩展:

pip install Flask-HTTPClient

参考文献

  1. 维基百科Flask
  2. flask扩展官方文档
  3. Flask-HTTPClient
  4. requests最佳实践
  5. 怎样将Python项目发布到PyPI
  6. pypi库Flask-HTTPClient
  7. Python 库打包分发(setup.py 编写)简易指南

`如果该文章对您产生了帮助,或者您对技术文章感兴趣,可以关注微信公众号: 技术茶话会, 能够第一时间收到相关的技术文章,谢谢!

Flask扩展之http客户端开发-LMLPHP`


本篇文章由一文多发平台ArtiPub自动发布
03-24 18:11