API Documentation

Let’s provide an example:

import asyncio
import peewee
import logging
from peewee_async import PostgresqlDatabase

database = PostgresqlDatabase('test')

# Disable sync queries
database.set_allow_sync(False)

# Let's define a simple model:
class PageBlock(peewee_async.AioModel):
    key = peewee.CharField(max_length=40, unique=True)
    text = peewee.TextField(default='')

    class Meta:
        database = database

– as you can see, nothing special in this code, just plain peewee_async.AioModel definition and disabling sync queries.

Now we need to create a table for model:

with database.allow_sync():
   PageBlock.create_table(True)

– this code is sync, and will do absolutely the same thing as would do code with regular peewee.PostgresqlDatabase. This is intentional, I believe there’s just no need to run database initialization code asynchronously! Less code, less errors.

Finally, let’s do something async:

async def my_async_func():
    # Add new page block
    await PageBlock.aio_create(
        key='title',
        text="Peewee is AWESOME with async!"
    )

    # Get one by key
    title = await PageBlock.aio_get(PageBlock, key='title')
    print("Was:", title.text)

    # Save with new text using manager
    title.text = "Peewee is SUPER awesome with async!"
    await title.aio_save()
    print("New:", title.text)

loop.run_until_complete(my_async_func())
loop.close()

That’s it! As you may notice there’s no need to connect and re-connect before executing async queries! It’s all automatic. But you can run AioDatabase.aio_connect() or AioDatabase.aio_close() when you need it.

And you can use methods from from AioModel for operations like selecting, deleting etc. All of them are listed below.

Databases

class peewee_async.databases.AioDatabase(database: str | None, **kwargs: Any)

async AioDatabase.aio_connect() → None: Creates a connection pool

property AioDatabase.is_connected: bool: Checks if pool is connected

async AioDatabase.aio_close() → None: Terminate pool backend. The pool is closed until you run aio_connect manually

async AioDatabase.aio_execute(query, fetch_results=None)

Execute SELECT, INSERT, UPDATE or DELETE query asyncronously.

Parameters:

query – peewee query instance created with Model.select(), Model.update() etc.
fetch_results – function with cursor param. It let you get data manually and don’t need to close cursor It will be closed automatically.

Returns:

result depends on query type, it’s the same as for sync query.execute()

AioDatabase.set_allow_sync(value: bool) → None: Allow or forbid sync queries for the database. See also the allow_sync() context manager.

AioDatabase.allow_sync() → Iterator[None]

Allow sync queries within context. Close sync connection on exit if connected.

Example:

with database.allow_sync():
    PageBlock.create_table(True)

AioDatabase.aio_atomic() → AsyncIterator[None]: Similar to peewee Database.atomic() method, but returns asynchronous context manager.

class peewee_async.PooledPostgresqlDatabase(database: str | None, **kwargs: Any)

PosgreSQL database driver providing single drop-in sync connection and async connections pool interface.

Parameters:: max_connections – connections pool size

Example:

database = PooledPostgresqlDatabase('test', max_connections=20)

class peewee_asyncext.PooledPostgresqlExtDatabase(database: str | None, **kwargs: Any)

PosgreSQL database extended driver providing single drop-in sync connection and async connections pool interface.

JSON fields support is always enabled, HStore supports is enabled by default, but can be disabled with register_hstore=False argument.

Parameters:: max_connections – connections pool size

Example:

database = PooledPostgresqlExtDatabase('test', register_hstore=False,
                                       max_connections=20)

class peewee_async.PooledMySQLDatabase(database: str | None, **kwargs: Any)

MySQL database driver providing single drop-in sync connection and async connections pool interface.

Parameters:: max_connections – connections pool size

Example:

database = PooledMySQLDatabase('test', max_connections=10)

AioModel

class peewee_async.AioModel(*args, **kwargs)

Async version of peewee.Model that allows to execute queries asynchronously with aio_execute method

Example:

class User(peewee_async.AioModel):
    username = peewee.CharField(max_length=40, unique=True)

await User.select().where(User.username == 'admin').aio_execute()

Also it provides async versions of peewee.Model shortcuts

Example:

user = await User.aio_get(User.username == 'user')

async classmethod AioModel.aio_get(*query, **filters)

Async version of peewee.Model.get

async classmethod AioModel.aio_get_or_none(*query, **filters)

Async version of peewee.Model.get_or_none

async classmethod AioModel.aio_create(**query)

Async version of peewee.Model.create

async classmethod AioModel.aio_get_or_create(**kwargs)

Async version of peewee.Model.get_or_create

async AioModel.aio_delete_instance(recursive=False, delete_nullable=False)

Async version of peewee.Model.delete_instance

async AioModel.aio_save(force_insert=False, only=None)

Async version of peewee.Model.save

async peewee_async.aio_prefetch(sq, *subqueries, prefetch_type)

Asynchronous version of prefetch().

AioModelSelect

class peewee_async.aio_model.AioModelSelect(model, fields_or_models, is_default=False): Async version of peewee.ModelSelect that provides async versions of ModelSelect methods

AioModelSelect.aio_scalar(database, as_tuple=False)

Get single value from select() query, i.e. for aggregation.

Returns:: result is the same as after sync query.scalar() call

async AioModelSelect.aio_get(database=None)

Async version of peewee.SelectBase.get

AioModelSelect.aio_count(database, clear_limit=False)

Async version of peewee.SelectBase.count

AioModelSelect.aio_exists(database)

Async version of peewee.SelectBase.exists

AioModelSelect.aio_prefetch(*subqueries, **kwargs)

Async version of peewee.ModelSelect.prefetch