peewee-aio 1.7.4

Description:

peeweeaio 1.7.4

Peewee-AIO
Async support for Peewee ORM



Features

Make Peewee ORM to work async
Supports PostgresQL, MySQL, SQLite
Supports asyncio and
trio
Contains types as well
Drivers supported:

aiosqlite
aiomysql
aiopg
asyncpg
triopg
trio_mysql



Requirements

python >= 3.8

Installation
peewee-aio should be installed using pip:
$ pip install peewee-aio

You can install optional database drivers with:
$ pip install peewee-aio[aiosqlite] # for SQLite (asyncio)
$ pip install peewee-aio[aiomysql] # for MySQL (asyncio)
$ pip install peewee-aio[aiopg] # for Postgresql (asyncio)
$ pip install peewee-aio[asyncpg] # for Postgresql (asyncio)
$ pip install peewee-aio[trio_mysql] # for MySQL (trio)
$ pip install peewee-aio[triopg] # for PostgresQL (trio)

Quickstart
import peewee
from peewee_aio import Manager, AIOModel, fields

manager = Manager('aiosqlite:///:memory:')

@manager.register
class Role(AIOModel):
# Pay attention that we are using fields from Peewee-AIO for better typing support
id = fields.AutoField()
name = fields.CharField()

@manager.register
class User(AIOModel):

# Pay attention that we are using fields from Peewee-AIO for better typing support
id = fields.AutoField()
name = fields.CharField()
role = fields.ForeignKeyField(Role)

async def handler():

# Initialize the database's pool (optional)
async with manager:

# Acquire a connection
async with manager.connection():

# Create the tables in database
await Role.create_table()
await User.create_table()

# Create a record
role = await Role.create(name='user')
assert role
assert role.id # role.id contains correct string type
user = await User.create(name="Andrey", role=role)
assert user
assert user.id
role = await user.role # Load role from DB using the foreign key
assert role # role has a correct Role Type

# Iterate through records
async for user in User.select(User, Role).join(Role):
assert user # user has a corrent User Type
assert user.id
role = await user.role # No DB query here, because the fk is preloaded

# Change records
user.name = "Dmitry"
await user.save()

# Update records
await User.update({"name": "Anonimous"}).where(User.id == user.id)

# Delete records
await User.delete().where(User.id == user.id)

# Drop the tables in database
await User.drop_table()
await Role.drop_table()

# Run the handler with your async library
import asyncio

asyncio.run(handler())

Usage
TODO
Sync usage
The library still supports sync mode (use manager.allow_sync):
class Test(peewee.Model):
data = peewee.CharField()

with manager.allow_sync():
Test.create_table()
Test.create(data='test')
assert Test.select().count()
Test.update(data='new-test').execute()

Get prefetched relations
TODO
# We prefetched roles here
async for user in User.select(User, Role).join(Role):
role = user.fetch(User.role) # get role from user relations cache

Bug tracker
If you have any suggestions, bug reports or annoyances please report them to
the issue tracker at https://github.com/klen/peewee-aio/issues
Contributing
Development of the project happens at: https://github.com/klen/peewee-aio
License
Licensed under a MIT License