API Documentation

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

Example:

database = Psycopg3Database(
    'database': 'postgres',
    'host': '127.0.0.1',
    'port': 5432,
    'password': 'postgres',
    'user': 'postgres',
    'pool_params': {
        "min_size": 0,
        "max_size": 5,
        'max_lifetime': 15
    }
)

See also: https://peewee.readthedocs.io/en/latest/peewee/api.html#Database

async AioDatabase.aio_connect() → None

Creates a connection pool

property AioDatabase.is_connected: bool

Checks if pool is connected

async AioDatabase.aio_close() → None

Close pool backend. The pool is closed until you run aio_connect manually.

async AioDatabase.aio_execute(query: Any, fetch_results: Callable[[AioDatabase, 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)

async AioDatabase.aio_begin() → Transaction

Start a new database transaction.

This method executes the SQL BEGIN statement and returns a Transaction object representing the started transaction.

Notes:

• This method must be called within an active aio_connection)() context manager.

• The returned Transaction() object should be used to manage commit or rollback operations.

Returns:

Transaction: An instance representing the active transaction.

async AioDatabase.aio_savepoint() → Transaction

Start a new transaction savepoint.

This method executes the SQL SAVEPOINT statement and returns a Transaction object representing the created savepoint.

Notes:

• This method must be called within an active aio_connection() context manager.

• The returned Transaction() object should be used to manage commit or rollback operations.

Returns:

Transaction: An instance representing the active savepoint.

AioDatabase.aio_atomic() → AbstractAsyncContextManager[None]

Create an async context-manager which runs any queries in the wrapped block in a transaction (or save-point if blocks are nested). Calls to aio_atomic() can be nested.

AioDatabase.aio_transaction() → AbstractAsyncContextManager[None]

Create an async context-manager that runs all queries in the wrapped block in a transaction.

Calls to aio_transaction() cannot be nested. If so OperationalError will be raised.

async AioDatabase.aio_create_tables(models: list[Any], **options: Any) → None

Async version of peewee.Database.create_tables https://docs.peewee-orm.com/en/4.0.0/peewee/api.html#Database.create_tables

async AioDatabase.aio_drop_tables(models: list[Any], **kwargs: Any) → None

Async version of peewee.Database.drop_tables https://docs.peewee-orm.com/en/4.0.0/peewee/api.html#Database.drop_tables

class peewee_async.Psycopg3Database(*args: Any, **kwargs: Any)

Extension for playhouse.Psycopg3Database providing extra methods for managing async connection based on psycopg3 pool backend.

Example:

database = Psycopg3Database(
    'database': 'postgres',
    'host': '127.0.0.1',
    'port': 5432,
    'password': 'postgres',
    'user': 'postgres',
    'pool_params': {
        "min_size": 0,
        "max_size": 5,
        'max_lifetime': 15
    }
)

See also: https://docs.peewee-orm.com/en/4.0.0/peewee/api.html#PostgresqlDatabase https://www.psycopg.org/psycopg3/docs/advanced/pool.html

class peewee_async.PostgresqlDatabase(*args: Any, **kwargs: Any)

Extension for playhouse.PostgresqlDatabase providing extra methods for managing async connection based on aiopg pool backend.

Example:

database = PostgresqlDatabase(
    'database': 'postgres',
    'host': '127.0.0.1',
    'port':5432,
    'password': 'postgres',
    'user': 'postgres',
    'pool_params': {
        "minsize": 0,
        "maxsize": 5,
        "timeout": 30,
        'pool_recycle': 1.5
    }
)

See also: https://docs.peewee-orm.com/en/4.0.0/peewee/api.html#PostgresqlDatabase https://aiopg.readthedocs.io/en/stable/

class peewee_async.MySQLDatabase(*args: Any, **kwargs: Any)

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

Example:

database = MySQLDatabase(
    'database': 'mysql',
    'host': '127.0.0.1',
    'port': 3306,
    'user': 'root',
    'password': 'mysql',
    'connect_timeout': 30,
    "pool_params": {
        "minsize": 0,
        "maxsize": 5,
        "pool_recycle": 2
    }
)

See also: https://docs.peewee-orm.com/en/4.0.0/peewee/api.html#MySQLDatabase https://aiomysql.readthedocs.io/en/stable/

class peewee_async.SqliteDatabase(*args: Any, **kwargs: Any)

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

Example:

database = SqliteDatabase(
    'database': 'db.sqlite'
)

See also: https://github.com/omnilib/aiosqlite

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_create_table(safe: bool = True, **options: Any) → None

Async version of peewee.Model.create_table https://docs.peewee-orm.com/en/4.0.0/peewee/api.html#Model.create_table

async classmethod AioModel.aio_drop_table(safe: bool = True, drop_sequences: bool = True, **options: Any) → None

Async version of peewee.Model.drop_table https://docs.peewee-orm.com/en/4.0.0/peewee/api.html#Model.drop_table

async classmethod AioModel.aio_truncate_table(**options: Any) → None

Async version of peewee.Model.truncate_table https://docs.peewee-orm.com/en/4.0.0/peewee/api.html#Model.truncate_table

async classmethod AioModel.aio_get(*query: Any, **filters: Any) → Self

Async version of peewee.Model.get

See also: http://docs.peewee-orm.com/en/4.0.0/peewee/api.html#Model.get

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

Async version of peewee.Model.get_or_none

See also: http://docs.peewee-orm.com/en/4.0.0/peewee/api.html#Model.get_or_none

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

Async version of peewee.Model.create

See also: http://docs.peewee-orm.com/en/4.0.0/peewee/api.html#Model.create

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

Async version of peewee.Model.get_or_create

See also: http://docs.peewee-orm.com/en/4.0.0/peewee/api.html#Model.get_or_create

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

Async version of peewee.Model.delete_instance

See also: http://docs.peewee-orm.com/en/4.0.0/peewee/api.html#Model.delete_instance

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

Async version of peewee.Model.save

See also: http://docs.peewee-orm.com/en/4.0.0/peewee/api.html#Model.save

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

Asynchronous version of prefetch().

See also: http://docs.peewee-orm.com/en/4.0.0/peewee/api.html#prefetch

AioModelSelect

class peewee_async.aio_model.AioModelSelect(model, fields_or_models, is_default=False)

Asynchronous version of peewee.ModelSelect that provides async versions of ModelSelect methods

AioModelSelect.aio_peek(database: AioDatabase, n: int = 1) → Any

Asynchronous version of peewee.SelectBase.peek

AioModelSelect.aio_scalar(database: AioDatabase, as_tuple: bool = False, as_dict: bool = False) → Any

Asynchronous version of peewee.SelectBase.scalar

AioModelSelect.aio_first(database: AioDatabase, n: int = 1) → Any

Asynchronous version of peewee.SelectBase.first

async AioModelSelect.aio_get(database: AioDatabase | None = None) → Any

Asynchronous version of peewee.SelectBase.get

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

Asynchronous version of peewee.SelectBase.count

AioModelSelect.aio_exists(database: AioDatabase) → bool

Asynchronous version of peewee.SelectBase.exists

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

Asynchronous version of peewee.ModelSelect.prefetch

TransactionTestCase

class peewee_async.testing.TransactionTestCase(database: AioDatabase)

Asynchronous context manager for simplifying ORM testing.

Resetting the database to a known state upon exit.

This class:

• Wraps the enclosed code block in a single transaction.

• Uses one shared connection for all queries within the context.

• Disables usage of aio_atomic and aio_transaction methods inside the managed block.

Example:

async with TransactionTestCase(database):
    await TestModel.aio_create(text="Test 1")
    assert await TestModel.select().aio_exists()

assert not await TestModel.select().aio_exists()