Как заставить SQLAlchemy в Tornado быть асинхронным?

Ответ 1

ORM плохо подходят для явного асинхронного программирования, то есть, когда программист должен создавать явные обратные вызовы в любое время, когда происходит то, что использует сетевой доступ. Основной причиной этого является то, что ORM широко используют ленивый шаблон загрузки, который более или менее несовместим с явным async. Код выглядит следующим образом:

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

на самом деле испустит два отдельных запроса - один, когда вы скажете first() чтобы загрузить строку, а следующий, когда вы говорите user.addresses, в том случае, .addresses коллекция .addresses уже не существует или была истекла. По сути, почти каждая строка кода, которая имеет дело с конструкциями ORM, может блокировать IO, поэтому вы будете иметь обширные спагетты обратного вызова в течение нескольких секунд - и, что еще хуже, подавляющее большинство этих строк кода фактически не блокируют IO, поэтому все накладные расходы на соединение обратных вызовов вместе для того, что в противном случае было бы простой операцией доступа к атрибутам, сделают вашу программу значительно менее эффективной.

Основная проблема с явными асинхронными моделями заключается в том, что они добавляют огромные накладные вызовы функции Python к сложным системам - не только на стороне пользователя, как у вас с ленивой загрузкой, но и на внутренней стороне, а также о том, как система обеспечивает абстракцию вокруг API базы данных Python (DBAPI). Для SQLAlchemy даже иметь базовую поддержку async будет налагать серьезное снижение производительности на подавляющее большинство программ, которые не используют асинхронные шаблоны, и даже те асинхронные программы, которые не очень параллельны. Рассмотрим SQLAlchemy, или любой другой уровень ORM или абстракции, может иметь следующий код:

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

Вышеприведенный код выполняет то, что кажется простой операцией, выполняющей инструкцию SQL в соединении. Но, используя полностью async DBAPI, как расширение async psycopg2, вышеуказанные коды блокируют IO как минимум три раза. Поэтому, чтобы написать вышеприведенный код в явном асинхронном стиле, даже если в нем нет механизма асинхронизации, и обратные вызовы фактически не блокируются, означает, что вышеупомянутый вызов внешней функции становится как минимум тремя вызовами функций, а не одним, не включая накладные расходы через явную асинхронную систему или вызовы DBAPI. Таким образом, простому приложению автоматически присваивается штраф в размере 3х накладных вызовов функции, окружающих простую абстракцию вокруг выполнения инструкции. И в Python, вызов вызова функции - это все.

По этим причинам я продолжаю быть менее чем взволнован по поводу шумихи вокруг явных систем async, по крайней мере, до такой степени, что некоторые люди, похоже, хотят все асинхронно для всего, например, доставляя веб-страницы (см. Node.js). Я бы рекомендовал вместо этого использовать неявные асинхронные системы, в первую очередь gevent, где вы получаете все неблокирующие преимущества IO асинхронной модели и ни одна из структурных подробностей/недостатков явных обратных вызовов. Я продолжаю пытаться понять примеры использования этих двух подходов, поэтому я озадачен привлекательностью явного асинхронного подхода как решения всех проблем, то есть, как вы видите, с node.js - мы используем языки сценариев в первое место, чтобы сократить сложность и сложность кода, а также явное async для простых вещей, таких как доставка веб-страниц, кажется, ничего не делает, кроме добавления шаблона, который также может быть автоматизирован gevent или аналогичным, если блокировка IO - даже такая проблема в (большое количество сайтов с большим объемом данных отлично справляется с синхронной моделью ввода-вывода). Системы, основанные на Gevent, доказали свою эффективность, и их популярность растет, поэтому, если вам нравится автоматизация кода, предоставляемая ORM, вы также можете использовать автоматизацию планирования async-IO, которую предоставляет система, подобная gevent.

Обновление: Ник Коглан указал на свою замечательную статью, посвященную явной или неявной асинхронизации, которая также должна быть прочитана здесь. И я также был обновлен до того, что pep-3156 теперь приветствует взаимодействие с gevent, отменив ранее заявленную незаинтересованность в gevent, в основном благодаря статье Ника. Поэтому в будущем я бы рекомендовал гибрид Tornado с использованием gevent для логики базы данных, как только будет доступна система интеграции этих подходов.

Ответ 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())

Ответ 3

Я использую торнадо с sqlalchemy следующим образом:

from tornado_mysql import pools
from sqlalchemy.sql import table, column, select, join
from sqlalchemy.dialects import postgresql, mysql

# from models import M, M2

t = table(...)
t2 = table(...)

xxx_id = 10

j = join(t, t2, t.c.t_id == t2.c.id)
s = select([t]).select_from(j).where(t.c.xxx == xxx_id)

sql_str = s.compile(dialect=mysql.dialect(),compile_kwargs={"literal_binds": True})


pool = pools.Pool(conn_data...)
cur = yield pool.execute(sql_str)
data = cur.fetchone()

В этом случае мы можем использовать модели sqlalchemy и инструменты sqlalchemy для конструктивных запросов.

Ответ 4

Не торнадо, но мы вроде сделали асинхронный SQLAlchemy в asyncio в GINO project:

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())

Он немного похож, но на самом деле совсем другой, чем ORM SQLAlchemy. Потому что мы использовали только часть ядра SQLAlchemy и создали простой ORM поверх него. Он использует asyncpg под ним, поэтому он только для PostgreSQL.

Обновление: GINO теперь поддерживает Торнадо, благодаря вкладу Владимира Гончарова. См. docs here