TornadoでSQLAlchemyを非同期にするには？

Question

TornadoのSQLAlchemyをasyncにする方法は？ async mongo exampleのMongoDBの例が見つかりましたが、motorのようなものは見つけられませんでした。SQLAlchemyです。誰もがtornado.gen（私は以下SQLAlchemy、私のハンドラがデータベースから読み込んで結果を返す瞬間、私はこの非同期にしたいと思います）を使用して実行するSQLAlchemyクエリを行う方法を知っていますか？

Answer 1

ORMは、明示的な非同期プログラミング、つまりネットワークアクセスを使用する何らかのものが発生したときに明示的なコールバックを生成する必要がある場合には適しません。これの主な理由は、ORMが明示的な非同期と多かれ少なかれ互換性のあるlazy loadingパターンを広範に使用していることです。このようなコード： - あなたはuser.addressesを言うとき、行、および次をロードするためにfirst()を言うとき.addressesコレクションが既に存在していないという場合には、1

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

は、実際には2つの別々のクエリを発行します、または期限切れになっています。本質的に、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文を実行し、簡単な操作であると思われるものを行います。しかし、psycopg2の非同期拡張のような完全非同期DBAPIを使用すると、上記のコードはIO上で少なくとも3回ブロックされます。したがって、非同期エンジンが使用されておらず、コールバックが実際にブロックされていない場合でも、上記のコードを明示的な非同期スタイルで記述すると、上記の外部関数呼び出しは、オーバーヘッドを含まない、少なくとも3つの関数呼び出し明示的な非同期システムまたはDBAPI自身によって呼び出されます。したがって、単純なアプリケーションには、ステートメント実行の単純な抽象化を取り巻く関数呼び出しオーバーヘッドの3倍のペナルティが自動的に与えられます。 Pythonではfunction call overhead is everythingです。

これらの理由から、私は、Webページを配信するようなものはすべて非同期で実行したいと思うほど、明示的な非同期システムを取り巻く誇大宣伝に興奮しています（node.js ）。暗黙の非同期システムを使用することをお勧めします。特に、geventは、非同期モデルのすべてのノンブロッキングIOメリットと、明示的コールバックの構造的冗長性/欠点のいずれも取得しません。私はこれらの2つのアプローチのユースケースを理解しようとしています。そのため、すべての問題の解決策として明示的な非同期アプローチが魅力的です。js - 冗長性とコードの複雑さを減らすためにスクリプト言語を使用しています。ウェブページの配信などの単純なものに対する明示的な非同期は、何もしませんが、 IOをブロックすることがそのような問題であっても（多量の大量のWebサイトが同期IOモデルでうまくいく場合） Geventベースのシステムは実績があり、人気が高まっているので、ORMが提供するコード自動化が好きな場合は、geventのようなシステムが提供するasync-IOスケジューリングの自動化を採用することもできます。

更新：Nick Coghlanは、great article on the subject of explicit vs. implicit asyncを指摘しました。これもまた、ここで読む必要があります。また、Nickの記事の主なおかげで、以前に述べたgeventの不利な点を逆にして、pep-3156 now welcomes interoperability with geventという事実に更新されました。だから将来、これらのアプローチを統合するシステムが利用可能になると、データベースロジックのためにgeventを使ってTornadoのハイブリッドを推奨します。

Answer 2

私はこれまで同じ問題を抱えていましたが、信頼できるAsync-MySQLライブラリが見つかりませんでした。しかし、Asyncio + Postgresを使ってクールな解決策があります。あなただけのボックスのうち、SQLAlchemyのサポートが付属していますaiopgライブラリを使用する必要があります。

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ので竜巻を使用しています：

 

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ツールを使用してクエリを構成します。

Answer 4

竜巻はないが、我々GINO projectでasyncioで作られたSQLAlchemyの非同期の並べ替え：それは少しのように見える

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は現在、Vladimir Goncharovの貢献により、トルネードをサポートしています。この回答の投稿から4年が経過しています。docs here