如何让Tornado中的SQLAlchemy异步？

Question

如何使SQLAlchemy中Tornado是async？我在async mongo示例中找到了MongoDB的示例,但我找不到类似motor的内容SQLAlchemy.有没有人知道如何SQLAlchemy执行查询tornado.gen(我在MySQL下面使用SQLAlchemy,目前我的处理程序从数据库读取并返回结果,我想使这个异步).

Answer 1

ORM非常适合显式异步编程,也就是说,程序员必须在任何使用网络访问的情况下产生显式回调.这样做的主要原因是ORM广泛使用延迟加载模式,这种模式或多或少与显式异步不兼容.代码看起来像这样:

user = Session.query(User).first()
print user.addresses

实际上会发出两个单独的查询 - 一个是当你说first()要加载一行时,另一个是你说的user.addresses,如果该.addresses集合尚未存在或已经过期.从本质上讲,几乎所有处理ORM结构的代码都可能会阻塞IO,所以你会在几秒钟之内进入广泛的回调意向 - 更糟糕的是,绝大多数代码行实际上不会阻塞IO,因此,将回调连接在一起以进行简单的属性访问操作的所有开销将使您的程序效率大大降低.

显式异步模型的一个主要问题是它们为复杂系统增加了巨大的Python函数调用开销 - 不仅在面向用户方面,就像延迟加载一样,而是在内部方面以及系统如何提供围绕Python数据库API(DBAPI).对于SQLAlchemy来说,即使拥有基本的异步支持,也会对绝大多数不使用异步模式的程序造成严重的性能损失,甚至是那些不是高度并发的异步程序.考虑SQLAlchemy或任何其他ORM或抽象层,可能具有如下代码:

def execute(connection, statement):
     cursor = connection.cursor()
     cursor.execute(statement)
     results = cursor.fetchall()
     cursor.close()
     return results

上面的代码执行似乎是一个简单的操作,在连接上执行SQL语句.但是使用完全异步DBAPI(如psycopg2的异步扩展),上面的代码至少会在IO上阻塞三次.所以用显式异步样式编写上面的代码,即使没有使用异步引擎并且回调实际上没有阻塞,也意味着上面的外部函数调用至少变成了三个函数调用,而不是一个,不包括强加的开销通过显式异步系统或DBAPI调用自身.因此,一个简单的应用程序会自动给出围绕语句执行的简单抽象的函数调用开销的3倍的惩罚.而在Python中,函数调用开销就是一切.

出于这些原因,我仍然对围绕显式异步系统的炒作感到兴奋,至少在一定程度上,有些人似乎想要对所有事情进行异步处理,例如提供网页(请参阅node.js).我建议使用隐式异步系统,最明显的是gevent,在这里你可以获得异步模型的所有非阻塞IO优势,而不是显式回调的结构冗余/缺点.我继续尝试理解这两种方法的用例,所以我对显式异步方法作为所有问题的解决方案的吸引力感到困惑,就像你在node.js中看到的那样 - 我们正在使用脚本语言减少详细程度和代码复杂性的第一个地方,以及提供网页这样的简单事物的显式异步似乎什么都不做,只是添加样板,如果阻塞IO甚至是一个问题,那么也可以通过gevent或类似方法实现自动化.这样的情况(大量的高容量网站使用同步IO模型做得很好).基于Gevent的系统经过生产验证,并且它们的受欢迎程度正在增长,因此如果您喜欢ORM提供的代码自动化,您可能还希望采用像gevent这样的系统提供的async-IO-scheduling自动化.

更新:Nick Coghlan指出了他关于显式与隐式异步主题的伟大文章,这也是必须阅读的内容.而且我也更新了pep-3156现在欢迎与gevent的互操作性,扭转其先前所说的gevent不感兴趣的事实,这在很大程度上要归功于Nick的文章.因此,一旦集成了这些方法的系统可用,我将推荐将Tornado混合使用gevent作为数据库逻辑.

Answer 2

我过去遇到过同样的问题,但是找不到可靠的Async-MySQL库.然而,使用Asyncio + Postgres有一个很酷的解决方案.您只需要使用aiopg库,它随附SQLAlchemy支持:

import asyncio
from aiopg.sa import create_engine
import sqlalchemy as sa


metadata = sa.MetaData()

tbl = sa.Table('tbl', metadata,
           sa.Column('id', sa.Integer, primary_key=True),
           sa.Column('val', sa.String(255)))

@asyncio.coroutine
def go():
    engine = yield from create_engine(user='aiopg',
                                      database='aiopg',
                                      host='127.0.0.1',
                                      password='passwd')

    with (yield from engine) as conn:
        yield from conn.execute(tbl.insert().values(val='abc'))

        res = yield from conn.execute(tbl.select().where(tbl.c.val=='abc'))
        for row in res:
            print(row.id, row.val)


loop = asyncio.get_event_loop()
loop.run_until_complete(go())

Answer 3

不是龙卷风，但是我们有点发SQLAlchemy的异步在ASYNCIO GINO项目：

import asyncio
from gino import Gino, enable_task_local
from sqlalchemy import Column, Integer, Unicode, cast

db = Gino()


class User(db.Model):
    __tablename__ = 'users'

    id = Column(Integer(), primary_key=True)
    nickname = Column(Unicode(), default='noname')


async def main():
    await db.create_pool('postgresql://localhost/gino')

    # Create object, `id` is assigned by database
    u1 = await User.create(nickname='fantix')
    print(u1.id, u1.nickname)  # 1 fantix

    # Retrieve the same row, as a different object
    u2 = await User.get(u1.id)
    print(u2.nickname)  # fantix

    # Update affects only database row and the operating object
    await u2.update(nickname='daisy')
    print(u2.nickname)  # daisy
    print(u1.nickname)  # fantix

    # Returns all user objects with "d" in their nicknames
    users = await User.query.where(User.nickname.contains('d')).gino.all()

    # Find one user object, None if not found
    user = await User.query.where(User.nickname == 'daisy').gino.first()

    # Execute complex statement and return command status
    status = await User.update.values(
        nickname='No.' + cast(User.id, Unicode),
    ).where(
        User.id > 10,
    ).gino.status()

    # Iterate over the results of a large query in a transaction as required
    async with db.transaction():
        async for u in User.query.order_by(User.id).gino.iterate():
            print(u.id, u.nickname)


loop = asyncio.get_event_loop()
enable_task_local(loop)
loop.run_until_complete(main())

它看起来有点像，但实际上与 SQLAlchemy ORM大不相同。因为我们只使用了 SQLAlchemy 核心的一部分，并在其上构建了一个简单的 ORM。它在下面使用asyncpg，因此它仅适用于 PostgreSQL。

更新：GINO 现在支持 Tornado，感谢 Vladimir Goncharov 的贡献。在此处查看文档