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

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 для логики базы данных, как только будет доступна система интеграции этих подходов.

У меня была такая же проблема в прошлом, и я не смог найти надежную библиотеку 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()) 

Я использую торнадо с 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, tct_id == t2.c.id) s = select([t]).select_from(j).where(tcxxx == 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()
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, tct_id == t2.c.id) s = select([t]).select_from(j).where(tcxxx == 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 для конструктивных запросов.

Не торнадо, но мы как- то сделали async 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 теперь поддерживает Торнадо, благодаря вкладу Владимира Гончарова. См. Документы здесь