当前位置：首页 > news >正文

ORM（Tortoise-ORM）操作

news 2025/10/21 14:33:01

ORM操作

ORM（Object-Relational Mapping）是将对象与关系数据库进行映射的技术，使得开发者可以使用面向对象的方式来操作数据库
Tortoise-ORM是一个流行的Python ORM库，支持异步操作，适合与FastAPI结合使用。

创建模型

以选课系统为例

from tortoise import Model
from tortoise import fieldsclass Student(Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=32, description="姓名")pwd = fields.CharField(max_length=32, description="密码")sno = fields.IntField(description="学号")#多对一的关系clazz = fields.ForeignKeyField("models.Clazz", related_name="students")#多对多的关系courses = fields.ManyToManyField("models.Course", related_name="students")class Course(Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=32, description="课程名称")#多对一的关系teacher = fields.Fields.ForeignKeyField("models.Teacher", related_name="courses")class Clazz(Model):name = fields.CharField(max_length=32, description="班级名称")class Teacher(Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=32, description="教师名称")pwd = fields.CharField(max_length=32, description="密码")tno = fields.IntField(description="教师编号")

aerich迁移工具

初始化配置

只使用一次

aerich init -t tortoise_config.TORTOISE_ORM

初始化数据库

只使用一次

aerich init-db

生成迁移文件

aerich migrate --name "init"

执行迁移

aerich upgrade

api接口和restful规范

api接口

应用程序编程接口（Application Programming Interface，API）

就是应用程序提供了一个操作数据的入口，这个入口可以是函数或类方法，也可以是一个URL地址或者一个网络地址，当客户端需要操作数据时，就可以通过调用这个入口来实现。
restful规范

之前写了，不再赘述

使用ORM实现增删改查

查询

@student_api.get("/")#异步意思着可以在等待数据库响应时处理其他请求
async def getAllStudents():# ORM 查询所有学生信息students = await Student.all()# QuerySet:[Student, Student, ...]for stu in students:print("name:", stu.name, "sno:", stu.sno)  # 过滤查询 filterstudentsFilter1 = await Student.filter(clazz_id = 1)for stu in studentsFilter1:print("过滤查询 name:", stu.name, "sno:", stu.sno)# 获取单个对象 get; 返回的是一个模型类对象studentsGet1 = await Student.get(id = 1)print("获取单个学生 name:", studentsGet1.name, "sno:", studentsGet1.sno)# 模糊查询studentsFilter2 = await Student.filter(sno__gt = 2001)for stu in studentsFilter2:print("模糊查询1 name:", stu.name, "sno:", stu.sno)studentsFilter3 = await Student.filter(sno__in = [2001,2003])for stu in studentsFilter3:print("模糊查询2 name:", stu.name, "sno:", stu.sno)# value 查询; 返回的是一个字典列表studentsValues = await Student.all().values("name","sno","clazz_id","clazz__name")print("values:", studentsValues)# 一对多查询和多对多查询zhao = await Student.get(name="zhao")print("zhao的班级:", await zhao.clazz.values("name"))print("zhao的课程:", await zhao.courses.all().values("name","addr","teacher__name"))return {"所有学生": studentsValues}

渲染模板查询

@student_api.get("/index.html")
async def getAllStudent(Request: Request):templates = Jinja2Templates(directory="templates")students = await Student.all()return templates.TemplateResponse("index.html", {"request": Request,"students": students})

新增

class StudentIn(BaseModel):name : strpwd : strsno : int#多对一的关系clazz_id : int#多对多的关系courses : list[int] = []@field_validator("sno")def sno_validator(cls, v):if v < 1000:raise ValueError("学号必须大于1000")return v@student_api.post("/")
async def addStudent(student_in: StudentIn):# 插入到数据库#方式一# student = Student(name=student_in.name,#         pwd=student_in.pwd,#         sno=student_in.sno,#         clazz_id=student_in.clazz_id)# await student.save()#方式二student = await Student.create(name=student_in.name,pwd=student_in.pwd,sno=student_in.sno,clazz_id=student_in.clazz_id)# 多对多关系的处理choose_courses = await Course.filter(id__in=student_in.courses)#id__in 生成 SQL语句中的 IN 查询await student.courses.add(*choose_courses)return {"student": student}

修改

@student_api.put("/{student_id}")
async def updateStudent(student_id: int, student_in: StudentIn):data = student_in.model_dump()courses = data.pop("courses")  # 移除多对多关系字段await Student.filter(id=student_id).update(**data)# 处理多对多关系的更新edit_student = await Student.get(id=student_id)await edit_student.courses.clear()  # 清除现有的多对多关系await edit_student.courses.add(*await Course.filter(id__in=courses))return {"操作": "更新学生信息",}

删除

from fastapi.exceptions import HTTPException
@student_api.delete("/{student_id}")
async def deleteStudent(student_id: int):delete_count = await Student.filter(id=student_id).delete()if not delete_count:raise HTTPException(status_code=404, detail=f"主键为{student_id}的学生不存在")return {"student_id": student_id,"delete_count": delete_count}