Implemented database adapters:
- [x] aiomysql
- [ ] aiopg
- [ ] sqlite
Currently 125 test cases have been ported from peewee, not all of them but constantly increases.
Simple Atomic operations (transactions) are also supported, but now well tested.
pip install aiopeewee # or conda install aiopeewee
from aiopeewee import AioModel, AioMySQLDatabase from peewee import CharField, TextField, DateTimeField from peewee import ForeignKeyField, PrimaryKeyField db = AioMySQLDatabase('test', host='127.0.0.1', port=3306, user='root', password='') class User(AioModel): username = CharField() class Meta: database = db class Blog(AioModel): user = ForeignKeyField(User) title = CharField(max_length=25) content = TextField(default='') pub_date = DateTimeField(null=True) pk = PrimaryKeyField() class Meta: database = db # create connection pool await db.connect(loop) # count await User.select().count() # async iteration on select query async for user in User.select(): print(user) # fetch all records as a list from a query in one pass users = await User.select() # insert user = await User.create(username='kszucs') # modify user.username = 'krisztian' await user.save() # async iteration on blog set [b.title async for b in user.blog_set.order_by(Blog.title)] # close connection pool await db.close() # see more in the tests
Note that AioManyToManyField must be used instead of ManyToMany.
from aiopeewee import AioManyToManyField class User(AioModel): username = CharField(unique=True) class Meta: database = db class Note(AioModel): text = TextField() users = AioManyToManyField(User) class Meta: database = db NoteUserThrough = Note.users.get_through_model() async for user in note.users: # do something with the users
Currently the only limitation I'm aware of immidiate setting of instance relation must be replaced with a method call:
# original, which is not supported charlie.notes = [n2, n3] # use instead await charlie.notes.set([n2, n3])
Converting to dict requires the asyncified version of model_to_dict
from aiopeewee import model_to_dict serialized = await model_to_dict(user)