ToB企服应用市场:ToB评测及商务社交产业平台

标题: Sanic，一个快如闪电的异步 Python Web 框架 [打印本页]
作者: 没腿的鸟    时间: 2024-10-12 16:29
标题: 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 特性


Sanic 安装
  1. pip install sanic -i https://pypi.doubanio.com/simple
复制代码
默认安装最新版，也可指定你必要的版本


二、Sanic 各功能测试

快速上手
先来快速构建一个简朴的 Python Web 应用：
  1. from sanic import Sanic
  2. from sanic.response import json
  3. from datetime import datetime
  4. import multiprocessing
  6. app = Sanic("SanicAPP")
  7. HOST = "localhost"
  8. PORT = 7777
  10. app.config.FALLBACK_ERROR_FORMAT = 'json'
  11. app.config.ACCESS_LOG = True
  13. async def get_datetime():
  14.     return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
  16. @app.route('/getdatetime')
  17. async def getdatetime(request):
  18.     return json({"now": await get_datetime(), 'server_name': request.server_name, 'path': request.path})
  21. if __name__ == "__main__":
  22.     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)，尽管从个人角度来说此模式可能不太利于后期开发，可读性也不太好，但还是必要学习一下的：
  1. from query_tag import Query
  3. q = Query()
  5. async def request_parse(request):
  6.     platform, chain, address = 'platform', 'chain', 'address'
  7.     if request.method == 'POST':
  8.         parameters = request.json
  9.         platform, chain, address = parameters['platform'], parameters['chain'], parameters['address']
  10.     elif request.method == 'GET':
  11.         parameters = request.args
  12.         platform, chain, address = parameters['platform'][0], parameters['chain'][0], parameters['address'][0]
  13.     print(f'请求参数为{platform}, {chain}, {address}')
  14.     return platform, chain, address
  16. @app.route('/tag', methods=['GET', 'POST'], version=1, version_prefix='/api/v')
  17. async def main(request):
  18.     platform, chain, address = await request_parse(request)
  19.     if platform == 'labelcloud':
  20.         if chain == 'eth':
  21.             addr = q.query_etherscan(address=address)
  22.             return json({'addr': addr})
  23.         elif chain == 'bsc':
  24.             addr = q.query_bscscan(address=address)
  25.             return json({'addr': addr})
  26.         elif chain == 'polygon':
  27.             addr = q.query_polygonscan(address=address)
  28.             return json({'addr': addr})
  29.         else:
  30.             json({'error': f'this chain no exists, available in [eth, bsc, polygon]'})
  31.     elif platform == 'oklink':
  32.         addr = q.query_oklink_com(chain=chain, address=address)
  33.         return json({'addr': addr})
  34.     else:
  35.         return json({'error': f'this platform no exists, available in [labelcloud, oklink]'})
复制代码
此处展示了一个简朴的 FBV 例子，该接口允许 GET 及 POST 请求，并为其界说了接口版本，一个请求 url 如下：
  1. http://127.0.0.1:7777/api/v1/tag?platform=labelcloud&chain=eth&address=0x9B9DBA51F809dd0F9E2607C458f23C1BD35Ab01b
复制代码
其实这些框架的语法都相差不大，而 Sanic 的一大上风就是支持异步，所以速率会快许多，并发请求量越大，其上风就越明显，掌握它，将成为你的进阶技能树！

CBV 模式
其意为“基于类的视图”(Class-based View)，此种模式使得代码的可读性大大增强，不仅可以提高开发效率，还利于后期维护，特别是一个项目由多个团队成员协同开发时每每选择该模式，我们将上面的 FBV 模式代码变为 CBV 模式的代码：
  1. from sanic.views import HTTPMethodView
  3. # CBV 模式
  4. class TagView(HTTPMethodView):
  5.     async def get(self, request):
  6.         parameters = request.args
  7.         platform, chain, address = parameters.get('platform', [''])[0], parameters.get('chain', [''])[0], parameters.get('address', [''])[0]
  8.         if platform == 'labelcloud':
  9.             if chain == 'eth':
  10.                 addr = q.query_etherscan(address=address)
  11.                 return json({'addr': addr})
  12.             elif chain == 'bsc':
  13.                 addr = q.query_bscscan(address=address)
  14.                 return json({'addr': addr})
  15.             elif chain == 'polygon':
  16.                 addr = q.query_polygonscan(address=address)
  17.                 return json({'addr': addr})
  18.             else:
  19.                 json({'error': f'this chain no exists, available in [eth, bsc, polygon]'})
  20.         elif platform == 'oklink':
  21.             addr = q.query_oklink_com(chain=chain, address=address)
  22.             return json({'addr': addr})
  23.         else:
  24.             return json({'error': f'this platform is no exists, available in [labelcloud, oklink]'})
  26.     async def post(self, request, name):
  27.         pass
  29. #把类视图添加进路由
  30. app.add_route(TagView.as_view(), '/tag', version=1, version_prefix='/api/v')
复制代码
类 TagView 继承了 sanic 中 views 模块的 HTTPMethodView 类，而类下面的方法即为对应请求类型的处置惩罚逻辑，末了用 app 的 add_route() 方法将该类作为视图添加进应用的路由中，代码已变得非常优雅！

OpenAPI 文档
你还可以将你的接口变成一份精致的文档，使得其他人更方便的阅读及明确你的接口，你只需安装其所属的扩展工具：
  1. pip install sanic-ext -i https://pypi.doubanio.com/simple
复制代码
接下来布局 API 文档：
  1. from sanic import (
  2.     exceptions,
  3.     Sanic,
  4. )
  5. from sanic.views import HTTPMethodView
  6. from sanic_ext import (
  7.     openapi,
  8.     Extend,
  9. )
  10. from spiders.query_tag import Query
  12. app = Sanic('TagAPI')
  13. Extend(app)
  15. class TagView(HTTPMethodView):
  16.     @openapi.definition(
  17.         description='This API can get tag for labelcloud or oklink by asynchronous request.',
  18.         parameter=[
  19.                     {
  20.                         "name": "platform",
  21.                         "in": "query",
  22.                         "type": "string",
  23.                         "description": "Platform (labelcloud or oklink)",
  24.                         "default": "labelcloud"
  25.                     },
  26.                     {
  27.                         "name": "chain",
  28.                         "in": "query",
  29.                         "type": "string",
  30.                         "description": "labelcloud including (eth, bsc, polygon), oklink including (eth, bsc, polygon, tron, btc, avalanche, arbitrum, optimism)",
  31.                         'default': 'eth'
  32.                     },
  33.                     {
  34.                         "name": "address",
  35.                         "in": "query",
  36.                         "type": "string",
  37.                         "description": "Blockchain address",
  38.                         'default': '0xB72eD8401892466Ea8aF528C1af1d0524bc5e105'
  39.                     }
  40.         ]
  41.     )
  42.     async def get(self, request):
  43.         q = Query()
  44.         parameters = request.args
  45.         platform, chain, address = parameters['platform'][0], parameters['chain'][0], parameters['address'][0]
  46.         if platform == 'labelcloud':
  47.             if chain == 'eth':
  48.                 addr = q.query_etherscan(address=address)
  49.                 return json({'data': addr})
  50.             elif chain == 'bsc':
  51.                 addr = q.query_bscscan(address=address)
  52.                 return json({'data': addr})
  53.             elif chain == 'polygon':
  54.                 addr = q.query_polygonscan(address=address)
  55.                 return json({'data': addr})
  56.             else:
  57.                 raise exceptions.SanicException(message=f'this chain no exists, available in [eth, bsc, polygon]')
  58.         elif platform == 'oklink':
  59.             addr = q.query_oklink_com(chain=chain, address=address)
  60.             return json({'data': addr})
  61.         else:
  62.             return exceptions.SanicException(message=f'this platform is no exists, available in [labelcloud, oklink]')
复制代码
从代码可看出，其 API 文档的设置是利用 sanic-ext 中的 openapi 和 Extend 模块，此中 openapi 采用装饰器的方式附加在请求类型方法上，我们界说了接口的描述和一些参数属性，包括网址参数名、参数类型、参数描述、参数默认值等

打开网址
  1. http://127.0.0.0:7777/docs/redoc
复制代码
你将瞥见默认的 Redoc 风格的 API文档：


打开网址
  1. 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：
  1. pip install tortoise-orm
复制代码
界说一个博客模子：
  1. from tortoise.models import Model
  2. from tortoise import fields
  3. from datetime import date
  5. class Blog(Model):
  6.     headline = fields.CharField(max_length=100)
  7.     author = fields.CharField(max_length=20, default='makerchen66')
  8.     pub_date = fields.DateField(default=date.today())
  9.     content = fields.TextField()
  10.    
  11.     def __str__(self):
  12.         return self.headline
  13.       
  14.     class Meta:
  15.         db_table = 'blog'
复制代码
然后必要初始化模子和数据库：
  1. from tortoise import Tortoise, run_async
  3. async def init():
  4.     await Tortoise.init(
  5.         db_url='sqlite://db.sqlite3',
  6.         modules={'models': ['app.models']}
  7.     )
  8.     # Generate the schema
  9.     await Tortoise.generate_schemas()
  11. run_async(init()
复制代码
末了就可利用模子：
  1. # Create instance by save
  2. blog = Blog(headline='震惊！某知名女明星竟然。。。', content='在一个风雨交加的夜晚，某国知名功夫宗师--马宝锅，正在练功室内锻炼绝技。。。', author='makerchen66', pub_date=date(2006, 3, 3))
  3. await blog.save()
  5. # Or by .create()
  6. await Blog.create(headline='震惊！某知名女明星竟然。。。', content='在一个风雨交加的夜晚，某国知名功夫宗师--马宝锅，正在练功室内锻炼绝技。。。', author='makerchen66', pub_date=date(2006, 3, 3))
  8. # Now search for a record
  9. queryResult = await Blog.filter(headline__contains='女明星').first()
  10. print(queryResult.author)
复制代码
更多 Tortoise ORM 利用教程可参考官网：
  1. https://tortoise.github.io/#tutorial
复制代码
Tortoise ORM Github 项目所在：
  1. https://github.com/tortoise/tortoise-orm
复制代码
接下来将爬虫项目和 Sanic 结合起来


三、爬虫 API 实战项目

接下来创建一个爬虫 API 实战项目，并且使路由模块、爬虫模块、视图模块、项目启动文件、配置文件独立


由于代码量较大，以下只展示部分核心文件和代码，视图模块中的 tag_views.py 文件：


爬虫模块中的 query_tag.py 文件：


路由模块中的 tag_routers.py 文件：
  1. from views.tag_view import TagView
  4. class TagRouter:
  5.     def load_router(self, app):
  6.         app.add_route(TagView.as_view(), '/tag', version=1, version_prefix='/api/v')
复制代码
server.py 文件：


项目架构已搭建好，今后可以不停扩充新模块和功能


四、服务器摆设及接口性能测试

利用 screen 工具挂载 Sanic 项目：
  1. screen -S sanic_api
复制代码
检察是否创建成功：
  1. screen -ls
复制代码
进入项目：
  1. screen -d -r sanic_api
复制代码
进入项目所在根目录，创建虚拟环境：
  1. virtualenv venv --python=python3.9.17
复制代码
进入虚拟环境：
  1. source venv/bin/activate
复制代码
安装项目所需的 Python 环境：
  1. pip install -r requirements.txt -i https://pypi.doubanio.com/simple
复制代码
末了启动项目：
  1. python server.py
复制代码
启动成功：


对接口进行测试，由于利用 asyncio 且 CPU 工作数量较多，故并发处置惩罚量较大，速率很快：
  1. http://服务器IP:7777/api/v1/tag?platform=labelcloud&chain=eth&address=0x9B9DBA51F809dd0F9E2607C458f23C1BD35Ab01b
复制代码
若要退出该 screen 项目，使它挂载在后台，可以利用快捷键【Ctrl+a+d】

删除该 screen 项目：
  1. screen -S -X sanic_screen_id quit
复制代码
如果对性能有更进一步的要求，可以和 Nginx Docker 等结合摆设。


五、作者Info

   Author：小鸿的摸鱼一样寻常，Goal：让编程更有趣！

专注于 Web开发、爬虫，游戏开发，数据分析、天然语言处置惩罚，AI等，期待你的关注，让我们一起发展、一起Coding！

版权说明：本文禁止抄袭、转载，侵权必究！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4