最小OpenAPI异步服务器应用程序
aio-openapi的Python项目详细描述
aio openapi
这个库是用于aiohttp的异步web中间件，用于为带有OpenAPIv 3的restapi提供服务 规范和可选的PostgreSql数据库。
目录
- Installation
- Development
- Features
- Web App
- OpenAPI Documentation
- Database Integration
- Websockets
- Environment Variables
安装
pip install aio-openapi
开发
克隆存储库并创建虚拟环境venv。
通过运行安装脚本安装依赖项
./dev/install.sh
运行测试
pytest --cov
功能
- 使用aiohttp 的异步Web路由
- 使用python进行数据验证、序列化和非序列化dataclasses
- OpenApiv 3自动文档
- SqlAlchemy表达式语言
- 与asyncpg 的异步数据库交互
- 使用alembic 进行迁移
- sqlalchemy表作为python数据类
- 支持click命令行界面
- 可选sentry中间件
网络应用程序
要创建openapi restful应用程序，请遵循此架构（让我们调用文件main.py）
fromopenapi.restimportrestdefcreate_app():returnrest(openapi=dict(title='A REST API',...),base_path='/v1',allowed_tags=[...],validate_docs=True,setup_app=setup_app,commands=[...])defsetup_app(app):app.router.add_routes(...)returnappif__name__=='__main__':create_app().main()
create_app函数通过调用rest函数创建aiohttp服务器应用程序。
此函数将cli映射项中的click命令添加到
支持openapi文档的路由文档。
setup_app函数用于实际设置自定义应用程序，通常是通过添加中间件、路由，
关闭回调、数据库集成等。
OpenAPI文档
这个库提供了创建符合openapi v 3的端点的工具，并且 自动记录它们。
testtests/example目录中的示例
fromaiohttpimportwebfromopenapi.db.pathimportSqlApiPathfromopenapi.specimportoproutes=web.RouteTableDef()@routes.view('/tasks')classTasksPath(SqlApiPath):""" --- summary: Create and query Tasks tags: - Task """table='tasks'@op(query_schema=TaskOrderableQuery,response_schema=[Task])asyncdefget(self):""" --- summary: Retrieve Tasks description: Retrieve a list of Tasks responses: 200: description: Authenticated tasks """paginated=awaitself.get_list()returnpaginated.json_response()@op(response_schema=Task,body_schema=TaskAdd)asyncdefpost(self):""" --- summary: Create a Task description: Create a new Task responses: 201: description: the task was successfully added 422: description: Failed validation """data=awaitself.create_one()returnself.json_response(data,status=201)
数据库集成
此库提供与asyncpg的集成，后者是一种高性能的异步
带有PostgreSql数据库的连接器。
要添加数据库扩展，只需在applicationonsetup_app函数中使用get_db函数：
fromopenapi.dbimportget_dbdefsetup_app(app):db=get_db(app)meta=db.metadata
这将启用数据库连接和命令行工具（其中大部分来自alembic）：
python main.py db --help
数据库容器位于db应用程序密钥：
app['db']
websockets
这个库提供了一个简单的分布式websocket实用程序来创建 WebSocket远程过程调用（RPC）和发布/订阅。
fromaiohttpimportwebfromopenapi.wsimportSocketsapp=web.Application()...app['web_sockets']=Sockets(app)
rpc协议
rpc协议具有以下传入消息的结构
{"id":"abc","method":"rpc_method_name","payload":{...}}
客户端使用id将请求与相应的响应链接起来。
rpc调用的响应是或成功
{"id":"abc","method":"rpc_method_name","response":{...}}
或错误
{
"id": "abc",
"method": "rpc_method_name":
"error": {
...
}
}
发布/订阅
订阅消息时，需要使用订阅{RPC处理程序的^ {CD12>} MIXIN。 信息格式：
{"channel":"channel_name","event":"event_name","data":{...}}
环境变量
库使用几个环境变量来支持测试和部署。
DATASTORE：postgresql连接字符串（与SqlAlchemy语法相同）DBPOOL_MIN_SIZE：数据库连接池的最小大小（默认为10）- {CD15}}：数据库连接池的最大大小（默认值为10）
欢迎加入QQ群-->： 979659372
