aio-peewee

DeprecationWarning

The package is deprecated. Please use peewee-aio instead.

---

aio-peewee -- Peewee support for async frameworks (Asyncio, Trio, Curio)

The library doesn't make peewee work async, but allows you to use Peewee with your asyncio based libraries correctly.

Features

Tasks Safety. The library tracks of the connection state using Task-local storage, making the Peewee Database object safe to use with multiple tasks inside a loop.
Async management of connections for Peewee Connections Pool

Contents

DeprecationWarning
Initialization
Async Connect
Connection Pooling
Database URL
ASGI Middleware
Curio

Requirements

python >= 3.8

Installation

aio-peewee should be installed using pip:

pip install aio-peewee

QuickStart

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

Usage

Initialization

from aiopeewee import PostgresqlDatabaseAsync, SqliteDatabaseAsync, MySQLDatabaseAsync, CockroachDatabaseAsync

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

Async Connect

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

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

Connection Pooling

from aiopeewee import PooledPostgresqlDatabaseAsync, PooledSqliteDatabaseAsync, PooledMySQLDatabaseAsync, PooledCockroachDatabaseAsync

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

Database 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 Middleware

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 uses contextvars to store db connections. So you have to enable contextvars for Curio: https://curio.readthedocs.io/en/latest/howto.html#how-do-you-use-contextvars

Bug tracker

If you have any suggestions, bug reports or annoyances please report them to the issue tracker at https://github.com/klen/aio-peewee/issues

Contributing

Development of the project happens at: https://github.com/klen/aio-peewee

License

Licensed under a MIT license.