【测开能力提升-fastapi框架】fastapi能力提升

1 ORM操作

注释：
fastapi缺少合适的ORM，官方推荐：sqlalchemy，但我们就不听官方的话，我们选择使用Tortoise ORM，因为他是支持异步的

1.1 tortoise ORM支持的数据库

PostgreSQL（使用asyncpg）
SQLite（使用aiosqlite）
Mysql、MariaDB（使用aipmysql或者asyncmy）

1.2 创建ORM模型

(示例: b站上的选课系统)
1、下载tortoise-orm

pip install tortoise-orm

2、models.py

from tortoise.models 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="学号")# 一对多关系chas = fields.ForeignKeyField("models.Clas", related_name="students")# 多对多courses = fields.ManyToManyField("models.Course", related_name="students")class Clas(Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=32, description="班级名称")teacher = fields.ForeignKeyField("models.Teacher", related_name="courses")class Course(Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=32, description="课程名称")class Teacher(Model):id = fields.IntField(pk=True)name = fields.CharField(max_length=32, description="教师名称")tno = fields.IntField(description="教师号")

1.3 ORM数据迁移

注释：
使用aerich迁移工具进行ORM数据迁移
3、main.py

import uvicorn
from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise
from settings import TORTOISE_ORMapp = FastAPI()register_tortoise(app=app, config=TORTOISE_ORM)if __name__ == "__main__":uvicorn.run("main:app", port=5050, reload=True)

4、TORTOISE_ORM配置信息
settings.py

TORTOISE_ORM = {"connections": {"default": {# 数据库配置(目前配置mysql)# "engine": "tortoise.backends.asyncpg""engine": "tortoise.backends.mysql","credentials": {"host": "127.0.0.1","port": "3306","user": "root","password": "111111","database": "htai","minsize": 1,"maxsize": 5,"charset": "utf8mb4","echo": True}},},# 迁移模型应用配置，后必须添加aerich.models，是serich默认配置"apps": {"models": {"models": ["models", "aerich.models"],"default_connection": "default",}},"use_tz": False,"timezone": "Asia/Shanghai"
}

5、安装aerich

pip install aerich

6、初始化配置信息，仅需执行一次

aerich init -t settings.TORTOISE_ORM # TORTOISE_ORM配置的位置

7、初始化数据库，仅需执行一次

aerich init-db

注释：更新模型并重新迁移
aerich migrate [–name]（标记修改操作） # aerich migrate --name add_column
升级：aerich upgrade
降级：aerich downgrade

1.4 restful规范

注释：
GET：查询
POST：添加
PUT：修改
DELETE：删除

api/student.py

from fastapi import APIRouterstudent_api = APIRouter()@student_api.get('/')
async def getAllStudent():return {"message": "查看所有的学生"}@student_api.post('/')
async def addStudent():return {"message": "添加一个学生"}@student_api.get('/{student_id}')
async def getOneStudent(student_id: int):return {"message": "查了id为{0}的学生".format(student_id)}@student_api.put('/{student_id}')
async def updateOneStudent(student_id: int):return {"message": "更新id为{0}的学生".format(student_id)}@student_api.delete("/{student_id}")
async def deleteOneStudent(student_id: int):return {"message": "删除一个id为{0}学生".format(student_id)}

import uvicorn
from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise
from settings import TORTOISE_ORM
from api.student import student_apiapp = FastAPI()
app.include_router(student_api, prefix="/student", tags=["选课系统的学生接口"])register_tortoise(app=app, config=TORTOISE_ORM)if __name__ == "__main__":uvicorn.run("main:app", port=5050, reload=True)

在这里插入图片描述

1.5 tortoise_orm查询语句操作

api/student.py

@student_api.get('/')
async def getAllStudent():# 1. 查询students表中所有数据，拿到一个Queryset类型数据student_1 = await Student.all()# 打印每一个学生的姓名和学号 (一定要异步化)for stu in student_1:print(stu.name, stu.sno)# 2. 过滤查询 filterstudent_2 = await Student.filter(name="rain")# 3. 筛选class_id为2的学生student_3 = await Student.filter(clas_id=2)# 4. 获取查询 get 和filter区别：filter返回的是Query_set，get返回的是模型类对象student_4 = await Student.get(id=6)# 5. 模糊查询：例：查询学号大于2001的所有学生student_5 = await Student.filter(sno__gt=2001)# 6. 例：取出学号是2001和2002的学生student_6 = await Student.filter(sno_in=[2001, 2002])# 7. 例：取出学号在2001到2005之间的学生student_7 = await Student.filter(sno_range=[2001, 2002])# 8. values查询 单个值student_8 = await Student.all().values("name")# 9. values查询 多个值student_9 = await Student.all().values("name", "sno")return {"message": "查看所有的学生"}

1.6 ORM响应页面数据

api/student.py

from fastapi import APIRouter, Request
from models import *
from fastapi.templating import Jinja2Templatesstudent_api = APIRouter()@student_api.get('/')
async def getAllStudent():templates = Jinja2Templates(directory="templates")# 1. 查询students表中所有数据，拿到一个Queryset类型数据student = await Student.all()return templates.TemplateResponse("index.html", {"request": Request,"student":student})@student_api.post('/')
async def addStudent():return {"message": "添加一个学生"}@student_api.get('/{student_id}')
async def getOneStudent(student_id: int):return {"message": "查了id为{0}的学生".format(student_id)}@student_api.put('/{student_id}')
async def updateOneStudent(student_id: int):return {"message": "更新id为{0}的学生".format(student_id)}@student_api.delete("/{student_id}")
async def deleteOneStudent(student_id: int):return {"message": "删除一个id为{0}学生".format(student_id)}

1.7 ORM添加语句操作

api/student.py

from typing import Listfrom fastapi import APIRouter, Request
from pydantic import BaseModel, field_validatorfrom models import *
from fastapi.templating import Jinja2Templatesstudent_api = APIRouter()@student_api.get('/')
async def getAllStudent():templates = Jinja2Templates(directory="templates")# 1. 查询students表中所有数据，拿到一个Queryset类型数据student = await Student.all()return templates.TemplateResponse("index.html", {"request": Request,"student":student})class StudentIn(BaseModel):id: intname: strsno: intclas_id: intcourses: List[int] = []@field_validator("name")def name_must_alpha(cls, value):assert value.isalpha(), "name must be alpha"return value@field_validator("sno")def sno_validate(cls, value):assert 1000<=value<=10000, "学号要在1000-10000范围内"return value@student_api.post('/')
async def addStudent(student_in: StudentIn):# 插入到数据库中# 方式一student = Student(name=student_in.name, pwd=student_in.pwd, sno=student_in.sno, clas_id=student_in.clas_id)await student.save()# 方式二student_1 = await Student.create(name=student_in.name, pwd=student_in.pwd, sno=student_in.sno, clas_id=student_in.clas_id)return {"message": student_1}@student_api.get('/{student_id}')
async def getOneStudent(student_id: int):return {"message": "查了id为{0}的学生".format(student_id)}@student_api.put('/{student_id}')
async def updateOneStudent(student_id: int):return {"message": "更新id为{0}的学生".format(student_id)}@student_api.delete("/{student_id}")
async def deleteOneStudent(student_id: int):return {"message": "删除一个id为{0}学生".format(student_id)}

1.8 ORM多对多添加语句操作

api/student.py

from typing import Listfrom fastapi import APIRouter, Request
from pydantic import BaseModel, field_validatorfrom models import *
from fastapi.templating import Jinja2Templatesstudent_api = APIRouter()@student_api.get('/')
async def getAllStudent():templates = Jinja2Templates(directory="templates")# 1. 查询students表中所有数据，拿到一个Queryset类型数据student = await Student.all()return templates.TemplateResponse("index.html", {"request": Request,"student":student})class StudentIn(BaseModel):id: intname: strsno: intclas_id: intcourses: List[int] = []@field_validator("name")def name_must_alpha(cls, value):assert value.isalpha(), "name must be alpha"return value@field_validator("sno")def sno_validate(cls, value):assert 1000<=value<=10000, "学号要在1000-10000范围内"return value@student_api.post('/')
async def addStudent(student_in: StudentIn):# 插入到数据库中# 方式一student = Student(name=student_in.name, pwd=student_in.pwd, sno=student_in.sno, clas_id=student_in.clas_id)await student.save()# 方式二student_1 = await Student.create(name=student_in.name, pwd=student_in.pwd, sno=student_in.sno, clas_id=student_in.clas_id)# 多对多的关系绑定choose_course = await Course.filter(id__in=student_in.courses)student.courses.add(*choose_course)return {"message": student_1}@student_api.get('/{student_id}')
async def getOneStudent(student_id: int):return {"message": "查了id为{0}的学生".format(student_id)}@student_api.put('/{student_id}')
async def updateOneStudent(student_id: int):return {"message": "更新id为{0}的学生".format(student_id)}@student_api.delete("/{student_id}")
async def deleteOneStudent(student_id: int):return {"message": "删除一个id为{0}学生".format(student_id)}

1.8 ORM一对多，多对多查询语句操作

api/student.py

from typing import Listfrom fastapi import APIRouter, Request
from pydantic import BaseModel, field_validatorfrom models import *
from fastapi.templating import Jinja2Templatesstudent_api = APIRouter()@student_api.get('/')
async def getAllStudent():templates = Jinja2Templates(directory="templates")# 1. 查询students表中所有数据，拿到一个Queryset类型数据student = await Student.all()# 查一个学生的一对多查询alvin = await Student.get(name='alvin')await alvin.clas.values("name")# 多对多查询student_8 = await alvin.courses.all()student_9 = await Student.all().values("name", "clas__name", "courses__name")# 查所有学生的一对多查询students_7 = await Student.all().values("name", "clas__name")return templates.TemplateResponse("index.html", {"request": Request,"student":student})class StudentIn(BaseModel):id: intname: strsno: intclas_id: intcourses: List[int] = []@field_validator("name")def name_must_alpha(cls, value):assert value.isalpha(), "name must be alpha"return value@field_validator("sno")def sno_validate(cls, value):assert 1000<=value<=10000, "学号要在1000-10000范围内"return value@student_api.post('/')
async def addStudent(student_in: StudentIn):# 插入到数据库中# 方式一student = Student(name=student_in.name, pwd=student_in.pwd, sno=student_in.sno, clas_id=student_in.clas_id)await student.save()# 方式二student_1 = await Student.create(name=student_in.name, pwd=student_in.pwd, sno=student_in.sno, clas_id=student_in.clas_id)# 多对多的关系绑定choose_course = await Course.filter(id__in=student_in.courses)student.courses.add(*choose_course)return {"message": student_1}@student_api.get('/{student_id}')
async def getOneStudent(student_id: int):return {"message": "查了id为{0}的学生".format(student_id)}@student_api.put('/{student_id}')
async def updateOneStudent(student_id: int):return {"message": "更新id为{0}的学生".format(student_id)}@student_api.delete("/{student_id}")
async def deleteOneStudent(student_id: int):return {"message": "删除一个id为{0}学生".format(student_id)}

1.9 ORM查询一个学生信息操作

api/student.py

from typing import Listfrom fastapi import APIRouter, Request
from pydantic import BaseModel, field_validatorfrom models import *
from fastapi.templating import Jinja2Templatesstudent_api = APIRouter()@student_api.get('/')
async def getAllStudent():templates = Jinja2Templates(directory="templates")# 1. 查询students表中所有数据，拿到一个Queryset类型数据student = await Student.all()# 查一个学生的一对多查询alvin = await Student.get(name='alvin')await alvin.clas.values("name")# 多对多查询student_8 = await alvin.courses.all()student_9 = await Student.all().values("name", "clas__name", "courses__name")# 查所有学生的一对多查询students_7 = await Student.all().values("name", "clas__name")return templates.TemplateResponse("index.html", {"request": Request,"student":student})class StudentIn(BaseModel):id: intname: strsno: intclas_id: intcourses: List[int] = []@field_validator("name")def name_must_alpha(cls, value):assert value.isalpha(), "name must be alpha"return value@field_validator("sno")def sno_validate(cls, value):assert 1000<=value<=10000, "学号要在1000-10000范围内"return value@student_api.post('/')
async def addStudent(student_in: StudentIn):# 插入到数据库中# 方式一student = Student(name=student_in.name, pwd=student_in.pwd, sno=student_in.sno, clas_id=student_in.clas_id)await student.save()# 方式二student_1 = await Student.create(name=student_in.name, pwd=student_in.pwd, sno=student_in.sno, clas_id=student_in.clas_id)# 多对多的关系绑定choose_course = await Course.filter(id__in=student_in.courses)student.courses.add(*choose_course)return {"message": student_1}@student_api.get('/{student_id}')
async def getOneStudent(student_id: int):student = await Student.get(id=student_id)return {"message": "查了id为{0}的学生".format(student_id), "student": student}@student_api.put('/{student_id}')
async def updateOneStudent(student_id: int):return {"message": "更新id为{0}的学生".format(student_id)}@student_api.delete("/{student_id}")
async def deleteOneStudent(student_id: int):return {"message": "删除一个id为{0}学生".format(student_id)}

1.9 ORM更新接口操作

api/student.py

from typing import Listfrom fastapi import APIRouter, Request
from pydantic import BaseModel, field_validatorfrom models import *
from fastapi.templating import Jinja2Templatesstudent_api = APIRouter()@student_api.get('/')
async def getAllStudent():templates = Jinja2Templates(directory="templates")# 1. 查询students表中所有数据，拿到一个Queryset类型数据student = await Student.all()# 查一个学生的一对多查询alvin = await Student.get(name='alvin')await alvin.clas.values("name")# 多对多查询student_8 = await alvin.courses.all()student_9 = await Student.all().values("name", "clas__name", "courses__name")# 查所有学生的一对多查询students_7 = await Student.all().values("name", "clas__name")return templates.TemplateResponse("index.html", {"request": Request,"student":student})class StudentIn(BaseModel):id: intname: strsno: intclas_id: intcourses: List[int] = []@field_validator("name")def name_must_alpha(cls, value):assert value.isalpha(), "name must be alpha"return value@field_validator("sno")def sno_validate(cls, value):assert 1000<=value<=10000, "学号要在1000-10000范围内"return value@student_api.post('/')
async def addStudent(student_in: StudentIn):# 插入到数据库中# 方式一student = Student(name=student_in.name, pwd=student_in.pwd, sno=student_in.sno, clas_id=student_in.clas_id)await student.save()# 方式二student_1 = await Student.create(name=student_in.name, pwd=student_in.pwd, sno=student_in.sno, clas_id=student_in.clas_id)# 多对多的关系绑定choose_course = await Course.filter(id__in=student_in.courses)student.courses.add(*choose_course)return {"message": student_1}@student_api.get('/{student_id}')
async def getOneStudent(student_id: int):student = await Student.get(id=student_id)return {"message": "查了id为{0}的学生".format(student_id), "student": student}@student_api.put('/{student_id}')
async def updateOneStudent(student_id: int, student_in: StudentIn):data = student_in.dict()courses = data.pop("courses")await Student.filter(id=student_id).update(**data)# 设置多对多的选修课edit_stu = Student.get(id=student_id)choose_courses = Course.filter(id__in=courses)await edit_stu.courses.clear()edit_stu.courses.add(*choose_courses)return {"message": "更新id为{0}的学生".format(student_id), "student": edit_stu}@student_api.delete("/{student_id}")
async def deleteOneStudent(student_id: int):return {"message": "删除一个id为{0}学生".format(student_id)}

1.10 ORM删除接口操作

api/student.py

from typing import Listfrom fastapi import APIRouter, Request
from pydantic import BaseModel, field_validatorfrom models import *
from fastapi.templating import Jinja2Templates
from fastapi.exceptions import HTTPExceptionstudent_api = APIRouter()@student_api.get('/')
async def getAllStudent():templates = Jinja2Templates(directory="templates")# 1. 查询students表中所有数据，拿到一个Queryset类型数据student = await Student.all()# 查一个学生的一对多查询alvin = await Student.get(name='alvin')await alvin.clas.values("name")# 多对多查询student_8 = await alvin.courses.all()student_9 = await Student.all().values("name", "clas__name", "courses__name")# 查所有学生的一对多查询students_7 = await Student.all().values("name", "clas__name")return templates.TemplateResponse("index.html", {"request": Request,"student":student})class StudentIn(BaseModel):id: intname: strsno: intclas_id: intcourses: List[int] = []@field_validator("name")def name_must_alpha(cls, value):assert value.isalpha(), "name must be alpha"return value@field_validator("sno")def sno_validate(cls, value):assert 1000<=value<=10000, "学号要在1000-10000范围内"return value@student_api.post('/')
async def addStudent(student_in: StudentIn):# 插入到数据库中# 方式一student = Student(name=student_in.name, pwd=student_in.pwd, sno=student_in.sno, clas_id=student_in.clas_id)await student.save()# 方式二student_1 = await Student.create(name=student_in.name, pwd=student_in.pwd, sno=student_in.sno, clas_id=student_in.clas_id)# 多对多的关系绑定choose_course = await Course.filter(id__in=student_in.courses)student.courses.add(*choose_course)return {"message": student_1}@student_api.get('/{student_id}')
async def getOneStudent(student_id: int):student = await Student.get(id=student_id)return {"message": "查了id为{0}的学生".format(student_id), "student": student}@student_api.put('/{student_id}')
async def updateOneStudent(student_id: int, student_in: StudentIn):data = student_in.dict()courses = data.pop("courses")await Student.filter(id=student_id).update(**data)# 设置多对多的选修课edit_stu = Student.get(id=student_id)choose_courses = Course.filter(id__in=courses)await edit_stu.courses.clear()edit_stu.courses.add(*choose_courses)return {"message": "更新id为{0}的学生".format(student_id), "student": edit_stu}@student_api.delete("/{student_id}")
async def deleteOneStudent(student_id: int):deleteCount = await Student.filter(id=student_id).delete()if not deleteCount:raise HTTPException(status_code=404, detail="主键为{0}的学生不存在".format(student_id))return {"message": "删除一个id为{0}学生".format(student_id)}