DOCSEARCHbyAlgolia
Submit Docs

peewee-async Documentation

repository·master·Indexed 20 days ago

https://github.com/05bit/peewee-async

An asynchronous interface for the Peewee ORM powered by asyncio, supporting Python 3.10+. It provides non-blocking database drivers for PostgreSQL (via aiopg or psycopg3), MySQL (via aiomysql), and SQLite (via aiosqlite). The library introduces AioModel and AioDatabase to provide asynchronous analogues to Peewee's synchronous methods, prefixed with aio_, while allowing hybrid use of sync and async operations.

Tokens
14K
Snippets
50
Records
72
Agent score
70%

What's inside peewee-async

  1. Overview of peewee-async

    master

    peewee-async is an asynchronous interface for the peewee ORM, powered by asyncio. It provides asynchronous analogues for Peewee's synchronous methods, using the aio_ prefix. It is designed as a drop-in replacement for sync code while allowing synchronous operations to remain synchronous.

    Key Features:

    • Python Support: Works on Python 3.10+.
    • Database Support:
      • PostgreSQL: via aiopg or psycopg3.
      • MySQL: via aiomysql.
      • Sqlite: via aiosqlite.
    • Functionality: Supports basic operations and transactions.

  3. How connection management works in peewee-async

    master

    By default, peewee-async manages connections automatically. Every time an asynchronous query is executed (e.g., await MyModel.aio_get(id=1)), the library internally uses an async context manager: async with database.aio_connection() as connection:.

    This internal mechanism follows these rules:

    1. It checks the connection_context ContextVar.
    2. If connection_context is not None, it uses the existing connection from that context.
    3. If connection_context is None, it acquires a new connection from the pool.
    4. Upon exiting the context, it releases the connection and sets connection_context to None.

    This means standard async queries are self-contained: they acquire a connection, run the query, and release it immediately.

    # Standard automatic connection management
await MyModel.aio_get(id=1)
# Connection is acquired, used, and released automatically

  4. Avoid transaction bugs when using asyncio.gather

    master

    Transactions in peewee-async are tied to the connection associated with the current asyncio.Task (via contextvars).

    Critical Warning: If you start a transaction in one task and then use asyncio.gather() to run multiple queries, those queries will run in separate tasks. Each new task will acquire its own connection, meaning they will not be part of the transaction started in the parent task. This can lead to unexpected behavior and data inconsistency.

    # THIS WILL NOT WORK AS EXPECTED:
# The queries inside gather run in separate tasks/connections
async with db.aio_atomic():
    await asyncio.gather(
        TestModel.aio_create(text='FOO1'), 
        TestModel.aio_create(text='FOO2'), 
        TestModel.aio_create(text='FOO3')
    )

  5. How asynchronous and synchronous operations work together

    master

    peewee-async provides an asynchronous interface for the Peewee ORM using asyncio. It works by providing asynchronous analogues to Peewee's sync methods, prefixed with aio_.

    Key behaviors:

    • Hybrid Support: You can use both sync and async methods. Sync code remains sync.
    • Sync Control: You can explicitly disable synchronous operations using database.set_allow_sync(False) to ensure your code stays asynchronous.
    • Temporary Sync Access: If sync operations are disabled, you can temporarily re-enable them using the with database.allow_sync(): context manager.
    • Model Definition: Use peewee_async.AioModel instead of peewee.Model to define models that support async operations.
    import asyncio
import peewee
import peewee_async

database = peewee_async.PooledPostgresqlDatabase(database='db_name', user='user', host='127.0.0.1', port='5432', password='password')

class TestModel(peewee_async.AioModel):
    text = peewee.CharField()
    class Meta:
        database = database

# Disabling sync to force async usage
database.set_allow_sync(False)

async def handler():
    # Using async methods
    await TestModel.aio_create(text="Async text")
    all_objects = await TestModel.select().aio_execute()
    for obj in all_objects:
        print(obj.text)

asyncio.run(handler())

# Re-enabling sync for cleanup
with database.allow_sync():
    TestModel.drop_table(True)

  7. Execute raw SQL for transactions

    master

    For advanced database features (like specific isolation levels), you can use aio_execute_sql to send raw SQL commands. Ensure you manage the connection lifecycle and execute the corresponding COMMIT or ROLLBACK commands manually.

    Note: A transaction must be executed entirely within a single connection.

    async with db.aio_connection() as connection:
    await db.aio_execute_sql(sql="begin isolation level repeatable read;")
    await TestModel.aio_create(text='FOO')
    try:
        await TestModel.aio_create(text='FOO')
    except:
        await db.aio_execute_sql(sql="ROLLBACK")
    else:
        await db.aio_execute_sql(sql="COMMIT")

  8. Manually manage transactions

    master

    If you require fine-grained control, you can manage transactions manually by acquiring a connection via aio_connection() and using aio_begin() to start the transaction. You must then explicitly call .commit() or .rollback() on the transaction object within a try/except/else block.

    async with db.aio_connection() as connection:
    tr = await db.aio_begin() # BEGIN
    await TestModel.aio_create(text='FOO')
    try:
        await TestModel.aio_create(text='FOO')
    except:
        await tr.rollback() # ROLLBACK
    else:
        await tr.commit() # COMMIT

  9. Run synchronous queries using allow_sync

    master

    If you need to perform synchronous operations (like database initialization or schema migrations) within an asynchronous application, use the database.allow_sync() context manager.

    Warning: Using synchronous queries in an async application is expensive because a new connection is opened and closed for every query, and long-running sync queries will block the event loop, preventing other coroutines from executing. It is recommended to use sync queries only for tests or single-threaded setup tasks.

    # Use this for table creation or migrations
with database.allow_sync():
   PageBlock.create_table(True)

  10. Install peewee-async

    master

    Install peewee-async using pip with the appropriate extra for your database backend:

    • PostgreSQL (aiopg): pip install peewee-async[postgresql]
    • PostgreSQL (psycopg3): pip install peewee-async[psycopg]
    • MySQL: pip install peewee-async[mysql]
    • SQLite: pip install peewee-async[sqlite]
    pip install peewee-async[postgresql]
# or
pip install peewee-async[psycopg]
# or
pip install peewee-async[mysql]
# or
pip install peewee-async[sqlite]

  12. Fallback to clear_tables for synchronous queries

    master

    If your tests involve synchronous Peewee queries (e.g., using .create() instead of .aio_create()), TransactionTestCase will not work. In these cases, you should use a manual cleanup fixture (e.g., clear_tables) to delete records from your models after the test runs.

    This approach ensures a clean state by explicitly deleting data rather than relying on transaction rollbacks, which are incompatible with synchronous execution in this context.

    import pytest
from typing import AsyncGenerator

@pytest.fixture
async def clear_tables() -> AsyncGenerator[None, None]:
    yield
    for model in all_your_models:
        await model.delete().aio_execute()

async def test_model_sync_created(clear_tables: None) -> None:
    # Use synchronous queries when clear_tables is provided
    TestModel.create(text="Test 1")
    assert TestModel.exists()