aio-peewee 0.4.1

DeprecationWarning

该软件包已弃用。 请使用 peewee-aio 代替。

—

aio-peewee – Peewee支持异步框架（Asyncio，Trio，Curio）

该库不是使Peewee异步工作，而是允许您正确地使用Peewee与基于asyncio的库。

pip install aio-peewee

from aiopeewee import db_url

db = db_url.connect('postgres+async://locahost:5432/database')

async def main(id=1):
    async with db:
        item = Model.get(Model.id == 1)

    return item.name

初始化

from aiopeewee import PostgresqlDatabaseAsync, SqliteDatabaseAsync, MySQLDatabaseAsync, CockroachDatabaseAsync

 db = PostgresqlDatabaseAsync('my_app', user='app', password='db_password', host='10.1.0.8', port=3306)

异步连接

# Manual
async def main():
     await db.connect_async()
     # ...
     await db.close_async()

 # Context manager
async def main():
     async with db:
         # ...

连接池

from aiopeewee import PooledPostgresqlDatabaseAsync, PooledSqliteDatabaseAsync, PooledMySQLDatabaseAsync, PooledCockroachDatabaseAsync

db = PooledPostgresqlDatabaseAsync('my_database', max_connections=8, stale_timeout=300, user='postgres')

数据库URL

from aiopeewee import db_url

 db0 = db_url.connect('cockroachdb+async://localhost/db', **db_params)
 db1 = db_url.connect('cockroachdb+pool+async://localhost/db', **db_params)
 db2 = db_url.connect('mysql+async://localhost/db', **db_params)
 db3 = db_url.connect('mysql+pool+async://localhost/db', **db_params)
 db4 = db_url.connect('postgres+async://localhost/db', **db_params)
 db5 = db_url.connect('postgres+pool+async://localhost/db', **db_params)
 db6 = db_url.connect('sqlite+async://localhost/db', **db_params)
 db7 = db_url.connect('sqlite+pool+async://localhost/db', **db_params)
 db8 = db_url.connect('sqliteexc+async://localhost/db', **db_params)
 db9 = db_url.connect('sqliteexc+pool+async://localhost/db', **db_params)

ASGI中间件

import datetime as dt

from asgi_tools import App
from aiopeewee import PeeweeASGIPlugin
import peewee as pw


db = PeeweeASGIPlugin(url='sqlite+async:///db.sqlite')


@db.register
class Visit(pw.Model):
    created = pw.DateTimeField(default=dt.datetime.utcnow())
    address = pw.CharField()


db.create_tables()


app = App()


@app.route('/')
async def visits_json(request):
    """Store the visit and load latest 10 visits."""
    Visit.create(address=request.client[0])
    return [{
        'id': v.id, 'address': v.address, 'timestamp': round(v.created.timestamp()),
    } for v in Visit.select().order_by(Visit.id.desc()).limit(10)]


app = db.middleware(app)

Curio

aio-peewee 使用 contextvars 存储数据库连接。因此您必须为Curio启用 contextvars： https://curio.readthedocs.io/en/latest/howto.html#how-do-you-use-contextvars