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 = PooledPostgresqlExtDatabase(
    '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
    }
)

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

Extension for peewee.PostgresqlDatabase providing extra methods for managing async connection based on psycopg3 pool backend.

Example:

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

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

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

Example:

database = PooledPostgresqlExtDatabase(
    '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
    }
)

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

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

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.

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(
    '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
    }
)

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): Asynchronous version of peewee.ModelSelect that provides async versions of ModelSelect methods

AioModelSelect.aio_peek(database: AioDatabase | None, n: int = 1) → Any: Asynchronous version of peewee.SelectBase.peek

AioModelSelect.aio_scalar(database: AioDatabase | None, as_tuple: bool = False, as_dict: bool = False) → Any: Asynchronous version of peewee.SelectBase.scalar

AioModelSelect.aio_first(database: AioDatabase | None, 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 | None, clear_limit: bool = False) → int: Asynchronous version of peewee.SelectBase.count

AioModelSelect.aio_exists(database: AioDatabase | None) → 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