大家好！我是爱摸鱼的小鸿，关注我，收看每期的编程干货。

本篇文章将详细介绍 Python 高性能 Web 异步框架 Sanic 的各功能，并通过实战将爬虫(Spiders)模块+视图(Views)模块+路由(Routers)模块+模型(Models)模块结合形成一个各模块独立、高性能、可读性高、可扩展性高、具有精美的接口文档、易于后期维护的爬虫 API 项目，并部署在 Ubuntu 服务器上供团队调用。

目录

一、Sanic 简介及特性二、Sanic 各功能测试三、爬虫 API 实战项目四、服务器部署及接口性能测试五、作者Info

一、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 Sanicfrom sanic.response import jsonfrom datetime import datetimeimport multiprocessingapp = Sanic("SanicAPP")HOST = "localhost"PORT = 7777app.config.FALLBACK_ERROR_FORMAT = 'json'app.config.ACCESS_LOG = Trueasync 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 Queryq = 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 HTTPMethodViewfrom sanic_ext import ( openapi, Extend,)from spiders.query_tag import Queryapp = 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 Modelfrom tortoise import fieldsfrom datetime import dateclass 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_asyncasync 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 saveblog = 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 recordqueryResult = 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 TagViewclass 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！

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