aiopg是一个用于从asyncio (PEP-3156/tulip) 框架访问 PostgreSQL 数据库的Python 开发库。它是Psycopg数据库驱动程序异步功能的封装。

Example

import asyncio  from aiopg.pool import create_pool    dsn = 'dbname=jetty user=nick password=1234 host=localhost port=5432'      @asyncio.coroutine  def test_select():      pool = yield from create_pool(dsn)        with (yield from pool) as conn:          cur = yield from conn.cursor()          yield from cur.execute('SELECT 1')          ret = yield from cur.fetchone()          assert ret == (1,), ret      asyncio.get_event_loop().run_until_complete(test_select())

Example of SQLAlchemy optional integration

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())          for row in res:              print(row.id, row.val)      asyncio.get_event_loop().run_until_complete(go())

Please use:

$ python3 runtests.py

for executing project's unittests

项目主页：http://www.open-open.com/lib/view/home/1418346045167