arstecnica.sqlalchemy.async 0.4

用法

基本上，该模块包装了一组最小的sa类（当前 Engine，Connection，Transaction和ResultProxy）进入 异步对应项，您对它们进行如下处理 示例：

from asyncio import coroutine
from arstecnica.sqlalchemy.async 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 arstecnica.sqlalchemy.async 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

测试

要运行单元测试，您应该：

1. 创建一个python虚拟环境并在 开发模式：

   python3 -m venv env
   source env/bin/activate
   python setup.py develop
   

2. 安装pytest和psycopg2或pymysql：

   pip install pytest psycopg2 pymysql
   

3. 创建测试数据库，例如使用createdb testdb

4. 使用环境变量执行py.testrunner 数据库的sa url:

   TEST_DB_URL="postgresql://localhost/testdb" py.test src
   TEST_DB_URL="mysql+pymysql://localhost/testdb" py.test src
   

更改