用于sqlalchemy的异步中间件
metapensiero.sqlalchemy.asyncio的Python项目详细描述
aiopg和aiomysql项目允许使用 ^分别是{a3}和MySQL，甚至是通过SQLAlchemy他们的 不过，这种方法并不理想，因为它们重新实现了 含有不良故障的SA低水平物质的数量。
基于Twisted的Alchimia方式要轻得多，即使可能 性能稍差，不会带来意外惊喜。
用法
基本上，该模块包装了一组最小的sa类（当前 Engine，Connection，Transaction和ResultProxy）进入 异步对应项，您对它们进行如下处理 示例：
from asyncio import coroutine
from metapensiero.sqlalchemy.asyncio import create_engine
@coroutine
def do_something(db_url):
engine = create_engine(db_url)
with (yield from engine.connect()) as conn:
with (yield from conn.begin()) as trans:
yield from conn.execute(users.insert()
.values(id=42, name="Async",))
res = yield from conn.execute(users.select()
.where(users.c.id == 42))
rows = yield from res.fetchall()
res = yield from conn.execute(users.delete()
.where(users.c.id == 42))
assert res.rowcount == 1
如果您使用的是Python3.5或更高版本，则应将上面的代码编写为：
from metapensiero.sqlalchemy.asyncio import create_engine
async def do_something(db_url):
engine = create_engine(db_url)
async with await engine.connect() as conn:
async with await conn.begin() as trans:
await conn.execute(users.insert()
.values(id=42, name="Async",))
res = await conn.execute(users.select()
.where(users.c.id == 42))
rows = await res.fetchall()
res = await conn.execute(users.delete()
.where(users.c.id == 42))
assert res.rowcount == 1
测试
要运行单元测试，您应该：
创建一个python虚拟环境并在 开发模式：
python3 -m venv env source env/bin/activate python setup.py develop
安装pytest、pytest-asyncio和psycopg2-binary或pymysql：
pip install pytest pytest-asyncio psycopg2-binary pymysql
创建测试数据库，例如使用createdb testdb
使用环境变量执行py.testrunner 数据库的sa url:
TEST_DB_URL="postgresql://localhost/testdb" py.test tests TEST_DB_URL="mysql+pymysql://localhost/testdb" py.test tests
更改
1.0（2018-07-01）
- 重命名为metapensero.sqlalchemy.asyncio
0.4（2015-09-25）
- 包装调整
0.3（2015-09-23）
- 支持Python3.5异步上下文管理器
0.2（2015-09-09）
- PyPI上的第一个（可用）分布
0.1（私人）
效果相当不错
0.0（私人）
最初的努力
欢迎加入QQ群-->： 979659372
