Sanic，一个快如闪电的异步 Python Web 框架

大家好，欢迎来到IT知识分享网。

一、Sanic 简介及特性

应用场景
如果你希望快速搭建一个小型的 API 项目，又对速度有非常大的需求，那 Sanic 无疑是你的天选框架，很哇塞的哟！

Sanic 特性

直接支持生产环境部署
高度可扩展
内置快速网络服务器
具有异步支持
使用 Redoc、Swagger 的 OpenAPI 文档
CORS 保护等

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)

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() 方法将该类作为视图添加进应用的路由中，代码已变得十分优雅！

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': '0xB72eDEa8aF528C1af1d0524bc5e105' } ] ) 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

http://127.0.0.0:7777/docs/swagger

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

三、爬虫 API 实战项目

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')

四、服务器部署及接口性能测试

使用 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

http://服务器IP:7777/api/v1/tag?platform=labelcloud&chain=eth&address=0x9B9DBA51F809dd0F9E2607C458f23C1BD35Ab01b

screen -S -X sanic_screen_id quit

五、作者Info

免责声明：本站所有文章内容,图片，视频等均是来源于用户投稿和互联网及文摘转载整编而成，不代表本站观点，不承担相关法律责任。其著作权各归其原作者或其出版社所有。如发现本站有涉嫌抄袭侵权/违法违规的内容,侵犯到您的权益，请在线联系站长,一经查实,本站将立刻删除。本文来自网络,若有侵权，请联系删除，如若转载，请注明出处：https://haidsoft.com/115480.html