Sanic，一个快如闪电的异步 Python Web 框架
大家好！我是爱摸鱼的小鸿，关注我，收看每期的编程干货。本篇文章将具体先容 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)
于生产环境启动运行：
https://i-blog.csdnimg.cn/blog_migrate/41d5b1e7e3655b730e52ca712fec2b43.png
该程序创建了一个可以访问当前时间的接口，并且利用异步支持，程序处置惩罚速率会更快，还进行了一些全局配置：开启访问日记、开启自动重载、CPU 工作数量为当前系统的 1/5（CPU 数量设置越多，并发处置惩罚速率越快）、将 404 页面以 json 格式返回等
访问成功示例：
https://i-blog.csdnimg.cn/blog_migrate/a8ab23b7faca6db82da4d047f023d500.png
访问失败示例：
https://i-blog.csdnimg.cn/blog_migrate/fc63b209481df1a293c10ea2087890b2.png
访问日记：
https://i-blog.csdnimg.cn/blog_migrate/4032b606de2fb1e8790167b47a07a0e9.png
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'], parameters['chain'], parameters['address']
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 '})
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 '})
此处展示了一个简朴的 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', ['']), parameters.get('chain', ['']), parameters.get('address', [''])
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 '})
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 '})
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'], parameters['chain'], parameters['address']
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 ')
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 ')
从代码可看出，其 API 文档的设置是利用 sanic-ext 中的 openapi 和 Extend 模块，此中 openapi 采用装饰器的方式附加在请求类型方法上，我们界说了接口的描述和一些参数属性，包括网址参数名、参数类型、参数描述、参数默认值等
打开网址
http://127.0.0.0:7777/docs/redoc
你将瞥见默认的 Redoc 风格的 API文档：
https://i-blog.csdnimg.cn/blog_migrate/9d45efebc4badc0b486ea5d16bb2c311.png
打开网址
http://127.0.0.0:7777/docs/swagger
你将瞥见 Swagger 风格的 API文档：
https://i-blog.csdnimg.cn/blog_migrate/99b01a3e0b7f8e3fd75aa1f292ba802e.png
Tortoise ORM
ORM（Object Relational Mapping），中文为“对象关系映射”，目标是为了会集数据模子和数据规则，确保安全地管理数据（提供对 SQL 注入的免疫力），还可以提高开发效率。
而 Tortoise ORM 是一个利用 asyncio 语法的 ORM，其灵感泉源于 Django 自带的 ORM，所以它的语法和 Django ORM 极其相似
那么 Sanic 为啥要选择 Tortoise ORM 作为最佳搭档呢？
起首 Tortoise 本身就是利用 asyncio 语法的，与 Sanic 一样，其次它的 API 设计既干净又实用，性能上也比其他 Python ORM 要好：
https://i-blog.csdnimg.cn/blog_migrate/9081a0a4dd93938989d7b243e4000d84.png
从上图来看，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 实战项目，并且使路由模块、爬虫模块、视图模块、项目启动文件、配置文件独立
https://i-blog.csdnimg.cn/blog_migrate/3eb760b0f72fed0f46714bbbfac88b3a.png
由于代码量较大，以下只展示部分核心文件和代码，视图模块中的 tag_views.py 文件：
https://i-blog.csdnimg.cn/blog_migrate/091493473609e13dc4e1603b4ae09fa0.png
爬虫模块中的 query_tag.py 文件：
https://i-blog.csdnimg.cn/blog_migrate/df424fff1f4ca23c71f69cbb2b3192fb.png
路由模块中的 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 文件：
https://i-blog.csdnimg.cn/blog_migrate/97f95ab7f4f11a8345ab9b80829c2059.png
项目架构已搭建好，今后可以不停扩充新模块和功能
四、服务器摆设及接口性能测试
利用 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
启动成功：
https://i-blog.csdnimg.cn/blog_migrate/1351a75343ba3c427ea5b1af2359634d.png
对接口进行测试，由于利用 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！
版权说明：本文禁止抄袭、转载，侵权必究！
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]