禹过留声

禹过留声

关注
发信
关注(28)粉丝(399)

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

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
Powered by LMLPHP ©2024 禹过留声 0.075314