郑文峰的博客 郑文峰的博客
首页
  • python之路
  • go之路
  • 其他
  • redis
  • mysql
  • docker
  • k8s
读书破万卷
周刊
关于
  • 导航 (opens new window)
  • 代码片段 (opens new window)
  • 收藏
  • 友链
  • 外部页面

    • 开往 (opens new window)
  • 索引

    • 分类
    • 标签
    • 归档
GitHub (opens new window)

zhengwenfeng

穷则变,变则通,通则久
首页
  • python之路
  • go之路
  • 其他
  • redis
  • mysql
  • docker
  • k8s
读书破万卷
周刊
关于
  • 导航 (opens new window)
  • 代码片段 (opens new window)
  • 收藏
  • 友链
  • 外部页面

    • 开往 (opens new window)
  • 索引

    • 分类
    • 标签
    • 归档
GitHub (opens new window)
  • python

    • 基础

    • 第三方库

    • django

    • flask

    • tornado

      • tornado 文件上传
      • tornado 使用jwt完成用户异步认证
      • tornado 用户密码 bcrypt加密
      • tornado 结合wtforms使用表单操作
      • tornado finish和write区别
      • tornado 使用peewee-async 完成异步orm数据库操作
        • 简介
        • 配置
        • 创建model
        • 增删改查
        • 连表查询
    • 其他

  • go

  • 其他

  • 编程
  • python
  • tornado
zhengwenfeng
2022-08-10
目录

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

# 简介

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")
1
2
3

在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()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 创建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
1
2
3
4
5
6
7
8
9
10

创建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="个人简介")
1
2
3
4
5
6
7

创建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="学科")
1
2
3
4
5

使用工具类创建表
在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()
1
2
3
4
5
6
7
8

# 增删改查

下面是增删改查的例子。

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)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

# 连表查询

在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)
1
2
3
4
5
6
7
8
9

使用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)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#python#tornado
上次更新: 2023/01/15, 15:47:48
tornado finish和write区别
python简单使用grpc

← tornado finish和write区别 python简单使用grpc→

最近更新
01
django rest_framework 分页
03-20
02
学习周刊-第03期-第09周
03-03
03
学习周刊-第02期-第08周
02-24
更多文章>
Theme by Vdoing | Copyright © 2022-2023 zhengwenfeng | MIT License
  • 跟随系统
  • 浅色模式
  • 深色模式
  • 阅读模式