muffin-peewee-aio — Asynchronous Peewee ORM integration for the Muffin framework.

• Python >= 3.10

Install via pip:

$ pip install muffin-peewee-aio

You can include an async database driver as needed:

$ pip install muffin-peewee-aio[aiosqlite]
$ pip install muffin-peewee-aio[aiopg]
$ pip install muffin-peewee-aio[asyncpg]
$ pip install muffin-peewee-aio[aiomysql]

from muffin import Application
from muffin_peewee import Plugin as Peewee

# Create the application
app = Application("example")

# Initialize the plugin
db = Peewee()
db.setup(app, PEEWEE_CONNECTION="postgresql://postgres:postgres@localhost:5432/database")

# Or: db = Peewee(app, **options)

You can provide options either at initialization or through the application config.

You can also define options in the Application config using the PEEWEE_ prefix:

PEEWEE_CONNECTION = "postgresql://..."

Note: Muffin application config is case-insensitive.

Define your model:

class Test(db.Model):
    data = peewee.CharField()

Query the database:

@app.route("/")
async def view(request):
    return [t.data async for t in Test.select()]

By default, connections and transactions are managed automatically. To manage them manually, disable the config flags and use context managers:

@app.route("/")
async def view(request):
    async with db.connection():
        async with db.transaction():
            # Perform DB operations here
            ...

You can configure read replicas via the REPLICAS option:

db.setup(
    app,
    PEEWEE_CONNECTION="postgresql://...",
    PEEWEE_REPLICAS=[
        "postgresql://replica1/...",
        "postgresql://replica2/...",
    ],
)

Use db.replica() to open a read-only connection to a replica backend:

@app.route("/")
async def view(request):
    async with db.replica():
        return [t.data async for t in Test.select()]

Create a migration:

$ muffin example:app peewee-create [NAME] [--auto]

Run migrations:

$ muffin example:app peewee-migrate [NAME] [--fake]

Rollback the latest migration:

$ muffin example:app peewee-rollback

List all migrations:

$ muffin example:app peewee-list

Clear migration history from the database:

$ muffin example:app peewee-clear

Merge all migrations into one:

$ muffin example:app peewee-merge

You can use the conftest() context manager to auto-manage schema setup and teardown during testing:

import pytest

@pytest.mark.asyncio
async def test_example():
    async with db.conftest():
        # Tables are created and dropped automatically
        ...

Found a bug or have a suggestion? Please open an issue at: https://github.com/klen/muffin-peewee-aio/issues

Development takes place at: https://github.com/klen/muffin-peewee-aio Pull requests are welcome!

• klen (Kirill Klenov)

This project is licensed under the MIT license.