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(*args: Any, **kwargs: Any)

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

Parameters:

pool_params – parameters that are passed to the pool
min_connections – min connections pool size. Alias for pool_params.minsize
max_connections – max connections pool size. Alias for pool_params.maxsize

Example:

database = PooledPostgresqlExtDatabase(
    'test',
    'min_connections': 1,
    'max_connections': 5,
    'pool_params': {"timeout": 30, 'pool_recycle': 1.5}
)

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: Any, fetch_results: Callable[[CursorProtocol], Awaitable[Any]] | None = None) → Any

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(*args: Any, **kwargs: Any)

Extension for peewee.PostgresqlDatabase providing extra methods for managing async connection.

class peewee_async.PooledPostgresqlExtDatabase(*args: Any, **kwargs: Any)

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

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

Example:

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

class peewee_async.PooledMySQLDatabase(*args: Any, **kwargs: Any)

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

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: Any, **filters: Any) → Self

Async version of peewee.Model.get

async classmethod AioModel.aio_get_or_none(*query: Any, **filters: Any) → Self | None

Async version of peewee.Model.get_or_none

async classmethod AioModel.aio_create(**query: Any) → Self

Async version of peewee.Model.create

async classmethod AioModel.aio_get_or_create(**kwargs: Any) → Tuple[Self, bool]

Async version of peewee.Model.get_or_create

async AioModel.aio_delete_instance(recursive: bool = False, delete_nullable: bool = False) → int

Async version of peewee.Model.delete_instance

async AioModel.aio_save(force_insert: bool = False, only: Any | None = None) → int

Async version of peewee.Model.save

async peewee_async.aio_prefetch(sq: Any, *subqueries: Any, prefetch_type: {'WHERE': 1, 'JOIN': 2} = 1) → Any

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: AioDatabase | None, as_tuple: bool = False) → Any

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: AioDatabase | None = None) → Any

Async version of peewee.SelectBase.get

AioModelSelect.aio_count(database: AioDatabase | None, clear_limit: bool = False) → int

Async version of peewee.SelectBase.count

AioModelSelect.aio_exists(database: AioDatabase | None) → bool

Async version of peewee.SelectBase.exists

AioModelSelect.aio_prefetch(*subqueries: Any, prefetch_type: {'WHERE': 1, 'JOIN': 2} = 1) → Any

Async version of peewee.ModelSelect.prefetch