首页 > Python资料 博客日记
Sanic,一个快如闪电的异步 Python Web 框架
2024-07-15 17:00:05Python资料围观84次
大家好!我是爱摸鱼的小鸿,关注我,收看每期的编程干货。
本篇文章将详细介绍 Python 高性能 Web 异步框架 Sanic 的各功能,并通过实战将爬虫(Spiders)模块+视图(Views)模块+路由(Routers)模块+模型(Models)模块结合形成一个各模块独立、高性能、可读性高、可扩展性高、具有精美的接口文档、易于后期维护的爬虫 API 项目,并部署在 Ubuntu 服务器上供团队调用。
一、Sanic 简介及特性
说到 Python Web 框架, 你可能会想到 Flask、Django、Tornado、FastAPI这些;而本文将向大家介绍另一个 Python Web 框架 —— Sanic。
它是一个 Python 3.8+ Web 服务器和 Web 框架,旨在快速运行。它允许使用 Python 3.5 中添加的 async/await 语法,这使您的代码非阻塞且快速。
应用场景
如果你希望快速搭建一个小型的 API 项目,又对速度有非常大的需求,那 Sanic 无疑是你的天选框架,很哇塞的哟!
Sanic 特性
- 直接支持生产环境部署
- 高度可扩展
- 内置快速网络服务器
- 具有异步支持
- 使用 Redoc、Swagger 的 OpenAPI 文档
- CORS 保护等
Sanic 安装
pip install sanic -i https://pypi.doubanio.com/simple
默认安装最新版,也可指定你需要的版本
二、Sanic 各功能测试
快速上手
先来快速构建一个简单的 Python Web 应用:
from sanic import Sanic
from sanic.response import json
from datetime import datetime
import multiprocessing
app = Sanic("SanicAPP")
HOST = "localhost"
PORT = 7777
app.config.FALLBACK_ERROR_FORMAT = 'json'
app.config.ACCESS_LOG = True
async def get_datetime():
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
@app.route('/getdatetime')
async def getdatetime(request):
return json({"now": await get_datetime(), 'server_name': request.server_name, 'path': request.path})
if __name__ == "__main__":
app.run(host=HOST, port=PORT, debug=False, auto_reload=True, workers=multiprocessing.cpu_count() // 5)
于生产环境启动运行:
该程序创建了一个可以访问当前时间的接口,并且使用异步支持,程序处理速度会更快,还进行了一些全局配置:开启访问日志、开启自动重载、CPU 工作数量为当前系统的 1/5(CPU 数量设置越多,并发处理速度越快)、将 404 页面以 json 格式返回等
访问成功示例:
访问失败示例:
访问日志:
FBV 模式
其意为“基于函数的视图”(Function-based View),尽管从个人角度来说此模式可能不太利于后期开发,可读性也不太好,但还是需要学习一下的:
from query_tag import Query
q = Query()
async def request_parse(request):
platform, chain, address = 'platform', 'chain', 'address'
if request.method == 'POST':
parameters = request.json
platform, chain, address = parameters['platform'], parameters['chain'], parameters['address']
elif request.method == 'GET':
parameters = request.args
platform, chain, address = parameters['platform'][0], parameters['chain'][0], parameters['address'][0]
print(f'请求参数为{platform}, {chain}, {address}')
return platform, chain, address
@app.route('/tag', methods=['GET', 'POST'], version=1, version_prefix='/api/v')
async def main(request):
platform, chain, address = await request_parse(request)
if platform == 'labelcloud':
if chain == 'eth':
addr = q.query_etherscan(address=address)
return json({'addr': addr})
elif chain == 'bsc':
addr = q.query_bscscan(address=address)
return json({'addr': addr})
elif chain == 'polygon':
addr = q.query_polygonscan(address=address)
return json({'addr': addr})
else:
json({'error': f'this chain no exists, available in [eth, bsc, polygon]'})
elif platform == 'oklink':
addr = q.query_oklink_com(chain=chain, address=address)
return json({'addr': addr})
else:
return json({'error': f'this platform no exists, available in [labelcloud, oklink]'})
此处展示了一个简单的 FBV 例子,该接口允许 GET 及 POST 请求,并为其定义了接口版本,一个请求 url 如下:
http://127.0.0.1:7777/api/v1/tag?platform=labelcloud&chain=eth&address=0x9B9DBA51F809dd0F9E2607C458f23C1BD35Ab01b
其实这些框架的语法都相差不大,而 Sanic 的一大优势就是支持异步,所以速度会快很多,并发请求量越大,其优势就越明显,掌握它,将成为你的进阶技能树!
CBV 模式
其意为“基于类的视图”(Class-based View),此种模式使得代码的可读性大大增强,不仅可以提高开发效率,还利于后期维护,特别是一个项目由多个团队成员协同开发时往往选择该模式,我们将上面的 FBV 模式代码变为 CBV 模式的代码:
from sanic.views import HTTPMethodView
# CBV 模式
class TagView(HTTPMethodView):
async def get(self, request):
parameters = request.args
platform, chain, address = parameters.get('platform', [''])[0], parameters.get('chain', [''])[0], parameters.get('address', [''])[0]
if platform == 'labelcloud':
if chain == 'eth':
addr = q.query_etherscan(address=address)
return json({'addr': addr})
elif chain == 'bsc':
addr = q.query_bscscan(address=address)
return json({'addr': addr})
elif chain == 'polygon':
addr = q.query_polygonscan(address=address)
return json({'addr': addr})
else:
json({'error': f'this chain no exists, available in [eth, bsc, polygon]'})
elif platform == 'oklink':
addr = q.query_oklink_com(chain=chain, address=address)
return json({'addr': addr})
else:
return json({'error': f'this platform is no exists, available in [labelcloud, oklink]'})
async def post(self, request, name):
pass
#把类视图添加进路由
app.add_route(TagView.as_view(), '/tag', version=1, version_prefix='/api/v')
类 TagView 继承了 sanic 中 views 模块的 HTTPMethodView 类,而类下面的方法即为对应请求类型的处理逻辑,最后用 app 的 add_route() 方法将该类作为视图添加进应用的路由中,代码已变得十分优雅!
OpenAPI 文档
你还可以将你的接口变成一份精美的文档,使得其他人更方便的阅读及理解你的接口,你只需安装其所属的扩展工具:
pip install sanic-ext -i https://pypi.doubanio.com/simple
接下来布局 API 文档:
from sanic import (
exceptions,
Sanic,
)
from sanic.views import HTTPMethodView
from sanic_ext import (
openapi,
Extend,
)
from spiders.query_tag import Query
app = Sanic('TagAPI')
Extend(app)
class TagView(HTTPMethodView):
@openapi.definition(
description='This API can get tag for labelcloud or oklink by asynchronous request.',
parameter=[
{
"name": "platform",
"in": "query",
"type": "string",
"description": "Platform (labelcloud or oklink)",
"default": "labelcloud"
},
{
"name": "chain",
"in": "query",
"type": "string",
"description": "labelcloud including (eth, bsc, polygon), oklink including (eth, bsc, polygon, tron, btc, avalanche, arbitrum, optimism)",
'default': 'eth'
},
{
"name": "address",
"in": "query",
"type": "string",
"description": "Blockchain address",
'default': '0xB72eD8401892466Ea8aF528C1af1d0524bc5e105'
}
]
)
async def get(self, request):
q = Query()
parameters = request.args
platform, chain, address = parameters['platform'][0], parameters['chain'][0], parameters['address'][0]
if platform == 'labelcloud':
if chain == 'eth':
addr = q.query_etherscan(address=address)
return json({'data': addr})
elif chain == 'bsc':
addr = q.query_bscscan(address=address)
return json({'data': addr})
elif chain == 'polygon':
addr = q.query_polygonscan(address=address)
return json({'data': addr})
else:
raise exceptions.SanicException(message=f'this chain no exists, available in [eth, bsc, polygon]')
elif platform == 'oklink':
addr = q.query_oklink_com(chain=chain, address=address)
return json({'data': addr})
else:
return exceptions.SanicException(message=f'this platform is no exists, available in [labelcloud, oklink]')
从代码可看出,其 API 文档的设置是利用 sanic-ext 中的 openapi 和 Extend 模块,其中 openapi 采用装饰器的方式附加在请求类型方法上,我们定义了接口的描述和一些参数属性,包括网址参数名、参数类型、参数描述、参数默认值等
打开网址
http://127.0.0.0:7777/docs/redoc
你将看见默认的 Redoc 风格的 API文档:
打开网址
http://127.0.0.0:7777/docs/swagger
你将看见 Swagger 风格的 API文档:
Tortoise ORM
ORM(Object Relational Mapping),中文为“对象关系映射”,目的是为了集中数据模型和数据规则,确保安全地管理数据(提供对 SQL 注入的免疫力),还可以提高开发效率。
而 Tortoise ORM 是一个使用 asyncio 语法的 ORM,其灵感来源于 Django 自带的 ORM,所以它的语法和 Django ORM 极其相似
那么 Sanic 为啥要选择 Tortoise ORM 作为最佳搭档呢?
首先 Tortoise 本身就是使用 asyncio 语法的,与 Sanic 一样,其次它的 API 设计既干净又实用,性能上也比其他 Python ORM 要好:
从上图来看,Tortoise ORM 在各方面的功能支持确实比较良好,目前支持的数据库有 MySQL、SQLite、Oracle、PostgreSQL 等
安装 Tortoise ORM:
pip install tortoise-orm
定义一个博客模型:
from tortoise.models import Model
from tortoise import fields
from datetime import date
class Blog(Model):
headline = fields.CharField(max_length=100)
author = fields.CharField(max_length=20, default='makerchen66')
pub_date = fields.DateField(default=date.today())
content = fields.TextField()
def __str__(self):
return self.headline
class Meta:
db_table = 'blog'
然后需要初始化模型和数据库:
from tortoise import Tortoise, run_async
async def init():
await Tortoise.init(
db_url='sqlite://db.sqlite3',
modules={'models': ['app.models']}
)
# Generate the schema
await Tortoise.generate_schemas()
run_async(init()
最后就可使用模型:
# Create instance by save
blog = Blog(headline='震惊!某知名女明星竟然。。。', content='在一个风雨交加的夜晚,某国知名功夫宗师--马宝锅,正在练功室内锻炼绝技。。。', author='makerchen66', pub_date=date(2006, 3, 3))
await blog.save()
# Or by .create()
await Blog.create(headline='震惊!某知名女明星竟然。。。', content='在一个风雨交加的夜晚,某国知名功夫宗师--马宝锅,正在练功室内锻炼绝技。。。', author='makerchen66', pub_date=date(2006, 3, 3))
# Now search for a record
queryResult = await Blog.filter(headline__contains='女明星').first()
print(queryResult.author)
更多 Tortoise ORM 使用教程可参考官网:
https://tortoise.github.io/#tutorial
Tortoise ORM Github 项目地址:
https://github.com/tortoise/tortoise-orm
接下来将爬虫项目和 Sanic 结合起来
三、爬虫 API 实战项目
接下来创建一个爬虫 API 实战项目,并且使路由模块、爬虫模块、视图模块、项目启动文件、配置文件独立
由于代码量较大,以下只展示部分核心文件和代码,视图模块中的 tag_views.py 文件:
爬虫模块中的 query_tag.py 文件:
路由模块中的 tag_routers.py 文件:
from views.tag_view import TagView
class TagRouter:
def load_router(self, app):
app.add_route(TagView.as_view(), '/tag', version=1, version_prefix='/api/v')
server.py 文件:
项目架构已搭建好,往后可以不断扩充新模块和功能
四、服务器部署及接口性能测试
使用 screen 工具挂载 Sanic 项目:
screen -S sanic_api
查看是否创建成功:
screen -ls
进入项目:
screen -d -r sanic_api
进入项目所在根目录,创建虚拟环境:
virtualenv venv --python=python3.9.17
进入虚拟环境:
source venv/bin/activate
安装项目所需的 Python 环境:
pip install -r requirements.txt -i https://pypi.doubanio.com/simple
最后启动项目:
python server.py
启动成功:
对接口进行测试,由于使用 asyncio 且 CPU 工作数量较多,故并发处理量较大,速度很快:
http://服务器IP:7777/api/v1/tag?platform=labelcloud&chain=eth&address=0x9B9DBA51F809dd0F9E2607C458f23C1BD35Ab01b
若要退出该 screen 项目,使它挂载在后台,可以使用快捷键【Ctrl+a+d】
删除该 screen 项目:
screen -S -X sanic_screen_id quit
如果对性能有更进一步的要求,可以和 Nginx Docker 等结合部署。
五、作者Info
Author:小鸿的摸鱼日常,Goal:让编程更有趣!
专注于 Web开发、爬虫,游戏开发,数据分析、自然语言处理,AI等,期待你的关注,让我们一起成长、一起Coding!
版权说明:本文禁止抄袭、转载,侵权必究!
标签:
相关文章
最新发布
- 【Python】简单的爬虫抓取
- Python SpeechRecognition库安装及使用中文识别Python SpeechRecognition库安装及使用中文识别
- 探索Pyro4:Python中的远程对象通信艺术
- jax 和 jaxlib 的 cuda 版本安装
- ⭐️怎么查看电脑Python版本(如何在Windows、Linux和macOS系统上查看Python版本的命令和方法)——《跟老吕学Python》
- python 实现convolution neural network卷积神经网络算法
- 【Python】PyWebIO 初体验:用 Python 写网页
- 【Python学习笔记】 第4章 介绍Python对象类型
- 计算机等级考试2级(Python)知识点整理
- 在 python 中调用 C/C++
点击排行
- 版本匹配指南:Numpy版本和Python版本的对应关系
- Python 可视化 web 神器:streamlit、Gradio、dash、nicegui;低代码 Python Web 框架:PyWebIO
- 版本匹配指南:PyTorch版本、torchvision 版本和Python版本的对应关系
- 相关性分析——Pearson相关系数+热力图(附data和Python完整代码)
- Could not build wheels for llama-cpp-python, which is required to install pyproject.toml-based proj
- Python pyinstaller打包exe最完整教程
- Anaconda版本和Python版本对应关系(持续更新...)
- Python与PyTorch的版本对应