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

郑文峰的博客

使用dify对接飞书多维表格使用n8n对接飞书多维表格服务启动时出现 OOM 一次服务升级时pg表DDL执行超时失败 Go语言高效IO缓冲技术详解 Go语言延迟初始化(Lazy Initialization)最佳实践 Go语言字符串拼接性能对比与优化指南 Go语言结构体内存对齐完全指南 Go语言空结构体：零内存消耗的高效编程 Go语言堆栈分配与逃逸分析深度解析 Go语言原子操作完全指南 Go语言内存预分配完全指南 Go语言不可变数据共享：无锁并发编程实践 Go语言零拷贝技术完全指南 Go语言遍历性能深度解析：从原理到优化实践 Go语言Interface Boxing原理与性能优化指南 Go协程池深度解析：原理、实现与最佳实践使用etcd分布式锁导致的协程泄露与死锁问题基于pre-commit的Python代码规范落地实践初识 MCP Server pulsar阻塞导致logstash无法接入日志 django-prometheus使用及源码分析 kube-proxy源码分析 kubernetes service如何通过iptables转发 tcp缓存引起的日志丢失 django-apschedule定时任务异常停止理解calico容器网络通信方案原理理解flannel的三种容器网络方案原理理解Linux IPIP隧道理解VXLAN网络理解Linux TunTap设备快速了解iptables kafka中listener和advertised.listeners的作用 django rest_framework 分页 django后端服务、logstash和flink接入VictoriaMetrics指标监控 python中import原理 docker容器单机网络手动实现docker容器bridge网络模型 mysql之MVCC原理 mysql之日志使用java开发logstash的filter插件使用python实现单例模式的三种方式 redis之缓存 redis之分片集群 redis之哨兵机制 redis之主从库同步 redis之持久化 redis之五种基本数据类型 go中如何处理error pod中将代码与运行环境分离 ddt源码分析 python装饰器的使用方法读书笔记:如何阅读一本书使用ddt实现unittest的参数化测试使用kubeadm安装k8s 优化gin表单的错误提示信息 gin中validator模块的源码分析 go简单使用grpc python简单使用grpc k8s之PV、PVC和StorageClass k8s之StatefulSet k8s之DaemonSet k8s之Job和CronJob k8s之ConfigMap和Secret k8s之Service k8s之Pod k8s之Deployment 容器的本质 docker容器 python迭代器与生成器 python元编程 python垃圾回收机制 python上下文管理器 django rest_framework使用jwt django rest_framework异常处理 django rest_framework 自定义文档 django压缩文件下载 django rest_framework使用pytest单元测试 django restframework choice 自定义输出数据 django Filtering 使用 django viewset 和 Router 配合使用时报的错 django model的序列化 django中使用AbStractUser django.core.exceptions.ImproperlyConfigured Application labels aren't unique, duplicates users django 中 media配置 django 外键引用自身和on_delete参数 django 警告 while time zone support is active Flask使用flask_socketio实现websocket flask结合mongo tornado 文件上传 tornado 使用jwt完成用户异步认证 tornado 用户密码 bcrypt加密 tornado 结合wtforms使用表单操作 tornado finish和write区别 pyspark streaming简介和消费 kafka示例使用hue创建ozzie的pyspark action workflow count的性能优化 django rest_framework Authentication django celery 结合使用网站

zhengwenfeng · 2022-08-10 · via 郑文峰的博客

# 简介

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

此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。原文来自 — 版权归原作者所有。

推荐订阅源

郑文峰的博客

# 简介

# 配置

# 创建model

# 增删改查

# 连表查询