社区首页 >专栏 >tornado 使用peewee-async 完成异步orm数据库操作

tornado 使用peewee-async 完成异步orm数据库操作

编程黑洞

发布于 2023-03-06 11:51:15

94900

代码可运行

文章被收录于专栏：编程黑洞编程黑洞

运行总次数：0

代码可运行

# 简介

tornado是一个异步web框架，其中不能使用阻塞的操作，不然会导致整个程序的阻塞。数据库操作时不可避免的需要使用，这里采用的是peewee-async去解决。

peewee-async 是一个为 peewee orm框架提供异步接口的库。

该项目的github地址: tornado_learning.git (opens new window)

# 配置

在settings.py文件中创建连接数据库 代码: server.py

import peewee_async

database = peewee_async.MySQLDatabase("tornado_learning", "127.0.0.1", port=3306, user="root", password="root1234")

在server.py中引用数据库连接，并加入到app中

from peewee_async import Manager
from tornado import web, ioloop

from tornado_learning.settings import database
from tornado_learning.settings import settings
from tornado_learning.urls import urlpattern


def make_app():    
    app = web.Application(urlpattern, debug=True, **settings)    
    
    # 就在这里添加数据库连接
    objects = Manager(database)    
    
    # 禁止使用同步操作
    database.set_allow_sync(False)    
    app.objects = objects    
    return app

if __name__ == '__main__':   
    app = make_app()   
    app.listen(8888)    
    ioloop.IOLoop.current().start()

# 创建model

创建通用的BaseModel类 create_time是每个model都需要的字段，将两个字段提取到BaseModel中。 id字段在peewee中会为每个model自动创建。为每一个model指定database 在配置目录tornado_learning中创建model.py

代码: tornado_learning/models

from datetime import datetime

from peewee import Model, DateTimeField

from tornado_learning.settings import database

class BaseModel(Model):   
    create_time = DateTimeField(default=datetime.now, verbose_name="创建时间")    
    class Meta:        
        database = database

创建model类

这里的student和teacher的关系是1对多。

创建student的模型类。代码: apps/school/models.py

from peewee import CharField, IntegerField, TextField
from tornado_learning.models import BaseModel

class Student(BaseModel):    
    name = CharField(max_length=100, null=False, verbose_name="学生名")    
    age = IntegerField(null=False, verbose_name="年龄")    
    desc = TextField(verbose_name="个人简介")

创建teacher的模型类

class Teacher(BaseModel):    
    student = ForeignKeyField(rel_model=Student, related_name="teachers")    
    name = CharField(max_length=100, null=False, verbose_name="老师名")   
    age = IntegerField(null=False, verbose_name="年龄")   
    subject = CharField(max_length=100, null=False, verbose_name="学科")

使用工具类创建表 在tools/init_db.py中初始化表。运行该文件即可在数据库中创建表

from tornado_learning.settings import database
from apps.school.models import Student

def init_db():    
    database.create_tables([Student, student])

if __name__ == '__main__':    
    init_db()

# 增删改查

下面是增删改查的例子。

form表单的使用可以参考我的文章<<tornado 结合wtforms使用表单操作

代码: apps/school/handler.py

import tornado

from apps.school.forms import StudentForm
from apps.school.models import Student
from tornado_learning.handler import BaseHandler

class StudentHandler(BaseHandler):

    async def get(self):
        id = self.get_argument("id", None)
        if not id:
            return self.write("please provide the 'id'")

        student = await self.application.objects.get(Student, id=id)

        try:
            self.write({
                "id": student.id,
                "name": student.name
            })
        except Student.DoesNotExist:
            raise tornado.webHttpError(404, "Object not found")

    async def post(self):

        student_form = StudentForm(self.request.arguments)
        if student_form.validate():
            await self.application.objects.create(Student, **student_form.data)

            self.write("创建成功")
        else:
            self.write("校验失败")

    async def delete(self):
        id = self.get_argument("id", None)
        if not id:
            return self.write("please provide the 'id'")

        student = await self.application.objects.get(Student, id=id)
        await self.application.objects.delete(student)

        self.write("删除成功")

    async def put(self):
        studentForm = StudentForm(self.request.arguments)

        student = Student(**studentForm.data)

        if studentForm.validate():
            await self.application.objects.update(student)
            self.write("更新成功")
        else:
            print(studentForm.errors)

# 连表查询

在teacher model中添加extend方法，拼凑连表查询的方法，方便使用。代码: apps/school/model.py

class Teacher(BaseModel):
    student = ForeignKeyField(rel_model=Student, related_name="teachers")
    name = CharField(max_length=100, null=False, verbose_name="老师名")
    age = IntegerField(null=False, verbose_name="年龄")
    subject = CharField(max_length=100, null=False, verbose_name="学科")

    @classmethod
    def extend(cls):
        return cls.select(cls, Student.name, Student.age).join(Student)

使用peewee拼凑出查询，然后通过异步执行得到结果代码: apps/school/handler.py

class TeacherHandler(BaseHandler):

    async def get(self):
        ret_data = {"data": []}

        teacher_query = Teacher.extend()
        teacher_query = teacher_query.filter(Teacher.age > 20)
        teachers = await self.application.objects.execute(teacher_query)

        for teacher in teachers:
            item_dict = {
                "teacher_name": teacher.name,
                "teacher_age": teacher.age,
                "student_name": teacher.student.name,
                "student_age": teacher.student.age
            }
            ret_data['data'].append(item_dict)

        return self.finish(ret_data)