Peewee Async

https://github.com/05bit/peewee-async
Admin
Peewee Async is an asynchronous interface for the peewee ORM powered by asyncio, providing...
Tokens:6,263
Snippets:85
Trust Score:6.7
License:MIT
Update:4 months ago
Show doc for...
Tokens:
Raw
### Install Dependencies

Source: https://github.com/05bit/peewee-async/blob/master/examples/README.md

Install the necessary Python dependencies for the examples.

```bash
pip install -r examples/requirements.txt
```

--------------------------------

### Peewee-async Quickstart Example

Source: https://github.com/05bit/peewee-async/blob/master/README.md

Demonstrates defining a model, performing synchronous operations, switching to asynchronous operations, and cleaning up. Ensure a 'test' PostgreSQL database is created before running.

```python
import asyncio
import peewee
import peewee_async

# Nothing special, just define model and database:

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

# Look, sync code is working!

TestModel.create_table(True)
TestModel.create(text="Yo, I can do it sync!")
database.close()

# No need for sync anymore!

database.set_allow_sync(False)

async def handler():
    await TestModel.aio_create(text="Not bad. Watch this, I'm async!")
    all_objects = await TestModel.select().aio_execute()
    for obj in all_objects:
        print(obj.text)

loop = asyncio.get_event_loop()
loop.run_until_complete(handler())
loop.close()

# Clean up, can do it sync again:
with database.allow_sync():
    TestModel.drop_table(True)

# Expected output:
# Yo, I can do it sync!
# Not bad. Watch this, I'm async!

```

--------------------------------

### Create Example Tables

Source: https://github.com/05bit/peewee-async/blob/master/examples/README.md

Run this command to create the necessary example tables in the database.

```bash
python -m examples.aiohttp_example
```

--------------------------------

### Install Requirements

Source: https://github.com/05bit/peewee-async/blob/master/load-testing/README.md

Use this command to install the necessary Python packages for the project.

```bash
pip install requirments
```

--------------------------------

### Start Docker Environment for Testing

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/installing.md

Start the Docker environment with a PostgreSQL database for running tests.

```console
docker-compose up -d
```

--------------------------------

### Start Database Service

Source: https://github.com/05bit/peewee-async/blob/master/examples/README.md

Use Docker Compose to start a PostgreSQL database service for development.

```bash
docker compose up postgres
```

--------------------------------

### Start Aiohttp Application

Source: https://github.com/05bit/peewee-async/blob/master/examples/README.md

Start the aiohttp application using Gunicorn with aiohttp.GunicornWebWorker.

```bash
gunicorn --bind 127.0.0.1:8080 --log-level INFO --access-logfile - \
      --worker-class aiohttp.GunicornWebWorker --reload \
      examples.aiohttp_example:app
```

--------------------------------

### Install Peewee-Async with SQLite

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/installing.md

Use this command to install the latest version of peewee-async with the SQLite backend.

```console
pip install peewee-async[sqlite]
```

--------------------------------

### Install Peewee-Async with MySQL

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/installing.md

Use this command to install the latest version of peewee-async with the MySQL backend.

```console
pip install peewee-async[mysql]
```

--------------------------------

### Run Uvicorn App

Source: https://github.com/05bit/peewee-async/blob/master/load-testing/README.md

Command to start the application server using uvicorn.

```bash
uvicorn app:app
```

--------------------------------

### Install Peewee-Async with PostgreSQL (aiopg)

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/installing.md

Use this command to install the latest version of peewee-async with the aiopg backend for PostgreSQL.

```console
pip install peewee-async[postgresql]
```

--------------------------------

### Install Development Dependencies

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/installing.md

Install the project's development dependencies using pip in editable mode.

```console
pip install -e .[dev]
```

--------------------------------

### Install Peewee-Async with PostgreSQL (psycopg3)

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/installing.md

Use this command to install the latest version of peewee-async with the psycopg3 backend for PostgreSQL.

```console
pip install peewee-async[psycopg]
```

--------------------------------

### Async Peewee with Tornado Setup and Handlers

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/examples.md

Sets up a Tornado application with peewee-async, defines a model, and implements request handlers for basic CRUD operations. Ensure the Tornado application is configured to run on an asyncio event loop.

```python
import tornado.web
import asyncio
import logging
import peewee
import peewee_async
from tornado.platform.asyncio import AsyncIOMainLoop

# Set up database and manager
database = peewee_async.PooledPostgresqlDatabase('test')


# Define model
class TestNameModel(peewee_async.AioModel):
    name = peewee.CharField()
    class Meta:
        database = database

    def __str__(self):
        return self.name


# Create table, add some instances
TestNameModel.create_table(True)
TestNameModel.get_or_create(id=1, defaults={'name': "TestNameModel id=1"})
TestNameModel.get_or_create(id=2, defaults={'name': "TestNameModel id=2"})
TestNameModel.get_or_create(id=3, defaults={'name': "TestNameModel id=3"})
database.close()

AsyncIOMainLoop().install()
app = tornado.web.Application(debug=True)
app.listen(port=8888)
app.database = database


# Add handlers
class RootHandler(tornado.web.RequestHandler):
    """Accepts GET and POST methods.

    POST: create new instance, `name` argument is required
    GET: get instance by id, `id` argument is required
    """
    async def post(self):
        name = self.get_argument('name')
        obj = await TestNameModel.aio_create(name=name)
        self.write({
            'id': obj.id,
            'name': obj.name
        })
  
    async def get(self):
        obj_id = self.get_argument('id', None)

        if not obj_id:
            self.write("Please provide the 'id' query argument, i.e. ?id=1")
            return

        try:
            obj = await TestNameModel.aio_get(id=obj_id)
            self.write({
                'id': obj.id,
                'name': obj.name,
            })
        except TestNameModel.DoesNotExist:
            raise tornado.web.HTTPError(404, "Object not found!")


class CreateHandler(tornado.web.RequestHandler):
    async def get(self):
        loop = asyncio.get_event_loop()
        task1 = asyncio.Task.current_task() # Just to demonstrate it's None
        task2 = loop.create_task(self.get_or_create())
        obj = await task2
        self.write({
            'task1': task1 and id(task1),
            'task2': task2 and id(task2),
            'obj': str(obj),
            'text': "'task1' should be null, "
                    "'task2' should be not null, "
                    "'obj' should be newly created object",
        })

    async def get_or_create(self):
        obj_id = self.get_argument('id', None)
        async with self.application.database.aio_atomic():
            obj, created = await TestNameModel.aio_get_or_create(
                id=obj_id,
                defaults={'name': "TestNameModel id=%s" % obj_id})
            return obj


app.add_handlers('', [
    (r"/", RootHandler),
    (r"/create", CreateHandler),
])

# Setup verbose logging
log = logging.getLogger('')
log.addHandler(logging.StreamHandler())
log.setLevel(logging.DEBUG)

# Run loop
print("""Run application server http://127.0.0.1:8888

    Try GET urls:
    http://127.0.0.1:8888?id=1
    http://127.0.0.1:8888/create?id=100

    Try POST with name=<some text> data:
    http://127.0.0.1:8888

^C to stop server""")
loop = asyncio.get_event_loop()
try:
    loop.run_forever()
except KeyboardInterrupt:
    print(" server stopped")

```

--------------------------------

### Perform Async CRUD Operations

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/quickstart.md

Execute asynchronous create, get, and save operations on the AioModel. The aio_create, aio_get, and aio_save methods handle database interactions without explicit connect/disconnect calls.

```python
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(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()
```

--------------------------------

### Manual Connection Management with Async Context Manager

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/connection.md

This shows how to manually manage database connections using `database.aio_connection` as an async context manager. This is useful for running multiple queries within a single connection.

```python
# acquire a connection and put it to connection_context context variable
async with database.aio_connection() as connection:

    # using the connection from the contextvar
    await MyModel.aio_get(id=1)

    # using the connection from the contextvar
    await MyModel.aio_get(id=2)

    # using the connection from the contextvar
    await MyModel.aio_get(id=3)
# release the connection set connection_context to None at the exit of the async contextmanager
```

--------------------------------

### Execute Query with Automatic Connection

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/connection.md

This demonstrates how a query is executed under an automatic async connection context manager. The connection is acquired, used, and then released.

```python
await MyModel.aio_get(id=1)
```

```python
async with database.aio_connection() as connection:
    await execute_your_query()
```

```python
# acquire a connection and put it to connection_context context variable
await MyModel.aio_get(id=1)
# release the connection and set the connection_context to None

# acquire a connection and put it to connection_context context variable
await MyModel.aio_get(id=2)
# release the connection and set the connection_context to None

# acquire a connection and put it to connection_context context variable
await MyModel.aio_get(id=3)
# release the connection and set the connection_context to None
```

--------------------------------

### Clone Peewee-Async Repository

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/installing.md

Clone the source code from the GitHub repository to set up a local development environment.

```console
git clone https://github.com/05bit/peewee-async.git
cd peewee-async
```

--------------------------------

### Create Table Synchronously

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/quickstart.md

Use the allow_sync() context manager to create database tables synchronously. This is intended for initialization and should not be used in performance-critical asynchronous code.

```python
with database.allow_sync():
   PageBlock.create_table(True)
```

--------------------------------

### Run Tests

Source: https://github.com/05bit/peewee-async/blob/master/README.md

Execute the test suite with pytest, showing verbose output and capturing logs.

```bash
pytest tests -v -s
```

--------------------------------

### Run Yandex.Tank Load Test

Source: https://github.com/05bit/peewee-async/blob/master/load-testing/README.md

Execute load tests using Dockerized Yandex.Tank. Mounts the current directory to /var/loadtest within the container.

```bash
docker run -v $(pwd):/var/loadtest --net host -it yandex/yandex-tank
```

--------------------------------

### Test Asynchronous Model Creation

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/testing.md

Demonstrates testing asynchronous model creation within a transaction. Changes made by this test will be rolled back automatically.

```python
async def test_model_created() -> None:
    # This test runs inside a transaction and is rolled back on exit.
    # No records will remain in the TestModel table after the test.
    await TestModel.aio_create(text="Test 1")

    assert await TestModel.aio_exists()
```

--------------------------------

### Run Tests with Pytest

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/installing.md

Execute the test suite using pytest with verbose output.

```console
pytest -s -v
```

--------------------------------

### Test Synchronous Model Creation with Table Clearing

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/testing.md

Illustrates testing synchronous model creation when `TransactionTestCase` cannot be used. This test relies on the `clear_tables` fixture to ensure a clean state.

```python
def test_model_sync_created(clear_tables: None) -> None:
    # TransactionTestCase cannot be used with synchronous queries,
    # so we use the clear_tables fixture instead.
    TestModel.create(text="Test 1")

    assert TestModel.exists()
```

--------------------------------

### Define Async Model with Pooled Database

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/quickstart.md

Define an AioModel and configure a PooledPostgresqlDatabase for asynchronous operations. Ensure to disable sync queries for optimal performance.

```python
import asyncio
import peewee
import logging
from peewee_async import PooledPostgresqlDatabase

database = PooledPostgresqlDatabase('test')


# 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
```

--------------------------------

### Async Test Fixtures for Transaction Management

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/testing.md

Sets up asynchronous fixtures for testing, including clearing tables and managing transactions. The `in_transaction` fixture uses `TransactionTestCase` for async tests and falls back to `clear_tables` for synchronous tests.

```python
import pytest
from peewee_async.testing import TransactionTestCase
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()


@pytest.fixture(autouse=True)
async def in_transaction(
    request: pytest.FixtureRequest,
) -> AsyncGenerator[None, None]:
    # In some cases, TransactionTestCase cannot be used,
    # so we fall back to the clear_tables fixture.
    if "clear_tables" in request.fixturenames:
        yield
    else:
        async with TransactionTestCase(database):
            yield
```

--------------------------------

### Manual Transaction Management

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/transaction.md

Manage transactions manually by acquiring a connection, beginning a transaction, and then explicitly committing or rolling back within a try-except-else block.

```python
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
```

--------------------------------

### Signal Support Overview

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/signals.md

This section details how to implement signal support in peewee-async. Models need to inherit from `peewee_async.signals.AioModel` to enable signal handling. Signal handlers, except for `pre_init`, must be asynchronous functions (coroutines).

```APIDOC
## Signal Support in Peewee-Async

### Description
Peewee-async offers signal support, enabling you to define handlers for various model events. To utilize signals, your models must inherit from `peewee_async.signals.AioModel`.

### Model Inheritance
Models intended to use signals must subclass `peewee_async.signals.AioModel`.

```python
from peewee_async.signals import AioModel

class MyModel(AioModel):
    # Model fields here
    pass
```

### Signal Handlers
Signal handlers are typically asynchronous functions (coroutines). The `pre_init` signal is an exception and must be a synchronous function.

### Available Signals

- `aio_pre_save`:
  Called immediately before an object is saved. Provides an additional keyword argument `created` (boolean) indicating if the model is new or being updated.

- `aio_post_save`:
  Called immediately after an object is saved. Provides an additional keyword argument `created` (boolean) indicating if the model is new or being updated.

- `aio_pre_delete`:
  Called immediately before an object is deleted using `Model.aio_delete_instance()`.

- `aio_post_delete`:
  Called immediately after an object is deleted using `Model.aio_delete_instance()`.

- `pre_init`:
  Called when a model class is first instantiated. This handler cannot be asynchronous.

### Example Usage

```python
from peewee_async.signals import AioModel, aio_post_save
from peewee import IntegerField

async def save_in_history_table(data):
    # Placeholder for history saving logic
    print(f"Saving data: {data}")

class MyModel(AioModel):
    data = IntegerField()

@aio_post_save(sender=MyModel)
async def on_save_handler(model_class, instance, created):
    await save_in_history_table(instance.data)
```
```

--------------------------------

### Define Async Model with Post-Save Signal Handler

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/signals.md

Models must subclass `peewee_async.signals.AioModel` to use signals. Signal handlers for `aio_post_save` should be coroutine functions.

```python
from peewee_async.signals import AioModel, aio_post_save


class MyModel(AioModel):
    data = IntegerField()

@aio_post_save(sender=MyModel)
async def on_save_handler(model_class, instance, created):
    await save_in_history_table(instance.data)
```

--------------------------------

### Test Aiohttp Application

Source: https://github.com/05bit/peewee-async/blob/master/examples/README.md

Send a curl request to the running aiohttp application to test its functionality.

```bash
curl 'http://127.0.0.1:8080/?p=1'
```

--------------------------------

### Raw SQL Transactions

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/transaction.md

Use raw SQL for transaction control, including setting isolation levels. Ensure that all SQL operations within the transaction use the same connection.

```python
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 await db.aio_execute_sql(sql="ROLLBACK")
    else:
        await await db.aio_execute_sql(sql="COMMIT")
```

--------------------------------

### Block PostgreSQL Connections

Source: https://github.com/05bit/peewee-async/blob/master/load-testing/README.md

Add a firewall rule to drop all incoming TCP traffic on port 5432, effectively blocking PostgreSQL connections.

```bash
sudo iptables -I INPUT -p tcp --dport 5432 -j DROP
```

--------------------------------

### Revert Firewall Rule

Source: https://github.com/05bit/peewee-async/blob/master/load-testing/README.md

Remove the previously added firewall rule that blocked PostgreSQL connections. Assumes the rule is the first one in the INPUT chain.

```bash
sudo iptables -D INPUT 1
```

--------------------------------

### Nested Atomic Transactions

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/transaction.md

Use `aio_atomic()` as a context manager for nested transactions. It automatically handles BEGIN, SAVEPOINT, RELEASE SAVEPOINT, and COMMIT/ROLLBACK based on nesting and exceptions.

```python
async with db.aio_atomic(): # BEGIN
    await TestModel.aio_create(text='FOO') # INSERT INTO "testmodel" ("text", "data") VALUES ('FOO', '') RETURNING "testmodel"."id"

    async with db.aio_atomic(): # SAVEPOINT PWASYNC__e83bf5fc118f4e28b0fbdac90ab857ca
        await TestModel.update(text="BAR").aio_execute() # UPDATE "testmodel" SET "text" = 'BAR'
    # RELEASE SAVEPOINT PWASYNC__e83bf5fc118f4e28b0fbdac90ab857ca
# COMMIT
```

--------------------------------

### Simple Atomic Transaction

Source: https://github.com/05bit/peewee-async/blob/master/docs/peewee_async/transaction.md

Use `aio_atomic()` for a single-level transaction. The block of code will be executed within a transaction that commits on success or rolls back on exception.

```python
async with db.aio_atomic(): # BEGIN
    await TestModel.aio_create(text='FOO') # INSERT INTO "testmodel" ("text", "data") VALUES ('FOO', '') RETURNING "testmodel"."id"
# COMMIT
```
Tokens:
Raw