python - ¿Cómo hacer que SQLAlchemy en Tornado sea asincrónico?

python-2.7

Question

¿Cómo hacer que SQLAlchemy en Tornado sea async ? Encontré un ejemplo para MongoDB en async mongo example pero no pude encontrar nada como motor para SQLAlchemy . ¿Alguien sabe cómo hacer que las consultas de SQLAlchemy ejecuten con tornado.gen (Estoy usando MySQL debajo de SQLAlchemy , en el momento en que mis manejadores leen desde la base de datos y devuelven el resultado, me gustaría hacer esto de manera asincrónica).

Answer 1

Estoy usando tornado con sqlalchemy de la siguiente manera:

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

En ese caso, podemos utilizar los modelos sqlalchemy y las herramientas sqlalchemy para construir consultas.

Answer 2

Los ORM son poco adecuados para una programación asincrónica explícita, es decir, donde el programador debe producir devoluciones de llamada explícitas cada vez que ocurre algo que utiliza acceso a la red. Una razón principal para esto es que los ORM hacen un uso extenso del patrón de carga diferida , que es más o menos incompatible con la asincrónica explícita. Código que se ve así:

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

en realidad emitirá dos consultas separadas, una cuando dices first() para cargar una fila, y la siguiente cuando dices user.addresses , en el caso de que la colección .addresses no esté presente o haya expirado. Esencialmente, casi todas las líneas de código que tratan con construcciones ORM pueden bloquearse en IO, por lo que estarás en espagueti extenso de devolución de llamada en cuestión de segundos, y para empeorar las cosas, la gran mayoría de esas líneas de código no se bloquearán en IO, de modo que todos los gastos generales de conectar devoluciones de llamadas juntos para lo que de otro modo serían simples operaciones de acceso a los atributos harán que su programa sea mucho menos eficiente también.

Un problema importante con los modelos asíncronos explícitos es que agregan una tremenda sobrecarga de funciones de Python a sistemas complejos, no solo en el lado del usuario como lo hace con la carga diferida, sino también en el lado interno con respecto a cómo el sistema proporciona abstracción alrededor del API de base de datos de Python (DBAPI). Para SQLAlchemy incluso tener soporte asíncrono básico impondría una grave penalización de rendimiento en la gran mayoría de los programas que no usan patrones asíncronos, e incluso aquellos programas asincrónicos que no son altamente concurrentes. Considera que SQLAlchemy, o cualquier otra capa de abstracción o ORM, podría tener un código como el siguiente:

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

El código anterior realiza lo que parece ser una operación simple, ejecutando una instrucción SQL en una conexión. Pero usando un DBAPI totalmente asincrónico como la extensión async de psycopg2, el código anterior bloquea en IO al menos tres veces. Entonces, para escribir el código anterior en estilo asíncrono explícito, incluso cuando no hay un motor asincrónico en uso y las devoluciones de llamada no son realmente bloqueantes, significa que la llamada de función externa anterior se convierte en al menos tres llamadas de función, en lugar de una, sin incluir la sobrecarga mediante el sistema asíncrono explícito o las llamadas DBAPI. Entonces, una aplicación simple recibe automáticamente una penalización de 3 veces la sobrecarga de llamada de función que rodea una simple abstracción alrededor de la ejecución de la declaración. Y en Python, la sobrecarga de llamada de función es todo .

Por estas razones, continúo menos que entusiasmado con la publicidad que rodea a los sistemas asíncronos explícitos, al menos en la medida en que algunas personas parecen querer sincronizar todo, como la entrega de páginas web (ver node.js). En su lugar, recomendaría utilizar sistemas asíncronos implícitos, sobre todo gevent , donde se obtienen todos los beneficios de IO no bloqueantes de un modelo asíncrono y ninguna de las verborreas / desventajas estructurales de las devoluciones de llamada explícitas. Continúo tratando de entender casos de uso para estos dos enfoques, por lo que estoy desconcertado por el atractivo del enfoque asíncrono explícito como una solución a todos los problemas, es decir, como se ve con node.js - estamos usando lenguajes de scripting en el primer lugar para reducir la verbosidad y la complejidad del código, y la asincr explícita para cosas simples como la entrega de páginas web parece no hacer más que agregar un texto estándar que puede ser automatizado por gevent o similar, si bloquear IO es incluso un problema en un caso como ese (muchos sitios web de alto volumen funcionan bien con un modelo IO sincrónico). Los sistemas basados ​​en Gevent están probados en producción y su popularidad está creciendo, así que si le gusta la automatización de códigos que brindan los ORM, también puede optar por la automatización async-IO-scheduling que proporciona un sistema como gevent.

Actualización : Nick Coghlan señaló su gran artículo sobre el tema de la asincría explícita frente a la implícita, que también es una lectura obligada aquí. Y también me han actualizado al hecho de que pep-3156 ahora da la bienvenida a la interoperabilidad con gevent , revirtiendo su desinterés previamente declarado en gevent, en gran parte gracias al artículo de Nick. Entonces, en el futuro recomendaría un híbrido de Tornado usando gevent para la lógica de la base de datos, una vez que el sistema de integración de estos enfoques esté disponible.

Answer 3

No es un tornado, pero hicimos SQLAlchemy asincronamente en asyncio en el proyecto 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())

Se parece un poco, pero en realidad es bastante diferente de SQLAlchemy ORM. Porque usamos solo una parte del núcleo de SQLAlchemy, y construimos un ORM simple encima. Utiliza asyncpg debajo, por lo que es solo para PostgreSQL .

Actualización : GINO ahora es compatible con Tornado, gracias a la contribución de Vladimir Goncharov. Ver documentos aquí

Answer 4

Tuve este mismo problema en el pasado y no pude encontrar una biblioteca Async-MySQL confiable. Sin embargo, hay una solución genial usando Asyncio + Postgres . Solo necesita usar la biblioteca aiopg , que viene con la compatibilidad SQLAlchemy lista para usar:

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