Flask扩展之http客户端开发

Posted 技术茶话会

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了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/


以上是关于Flask扩展之http客户端开发的主要内容,如果未能解决你的问题,请参考以下文章

Flask之邮件扩展

flask扩展系列之 - 访问速度限制

Flask之flask-script模块使用

87flask之flask-script组件

flask基础之扩展

Flask框架从入门到精通之扩展脚本(十五)