fastapi tortoise-orm 使用三
Last updated
Last updated
fastapi是一个很优秀的框架，但是缺少一个合适的orm，官方代码里面使用的是sqlalchemy，Tortoise ORM 是受 Django 启发的易于使用的异步 ORM （对象关系映射器）。
github文档: https://tortoise.github.io/
Tortoise ORM 目前支持以下数据库：
PostgreSQL >= 9.4（使用asyncpg）
SQLite（使用aiosqlite）
MySQL/MariaDB（使用aiomysql或使用asyncmy）
以选课系统为例：
models.py
from tortoise.models import Model
from tortoise import fields
class Clas(Model):
name = fields.CharField(max_length=255, description='班级名称')
main.py
import uvicorn
from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise
from settings import TORTOISE_ORM
app = FastAPI()
# 该方法会在fastapi启动时触发，内部通过传递进去的app对象，监听服务启动和终止事件
# 当检测到启动事件时，会初始化Tortoise对象，如果generate_schemas为True则还会进行数据库迁移
# 当检测到终止事件时，会关闭连接
register_tortoise(
app,
config=TORTOISE_ORM,
# generate_schemas=True, # 如果数据库为空，则自动生成对应表单，生产环境不要开
# add_exception_handlers=True, # 生产环境不要开，会泄露调试信息
)
if __name__ == '__main__':
uvicorn.run('main:app', host='127.0.0.1', port=8000, reload=True,
debug=True, workers=1)settings.py
TORTOISE_ORM = {
'connections': {
'default': {
# 'engine': 'tortoise.backends.asyncpg', PostgreSQL
'engine': 'tortoise.backends.mysql', # MySQL or Mariadb
'credentials': {
'host': '127.0.0.1',
'port': '3306',
'user': 'root',
'password': 'yuan0316',
'database': 'fastapi',
'minsize': 1,
'maxsize': 5,
'charset': 'utf8mb4',
"echo": True
}
},
},
'apps': {
'models': {
'models': ['apps.models', "aerich.models"],
'default_connection': 'default',
}
},
'use_tz': False,
'timezone': 'Asia/Shanghai'
}aerich是一种ORM迁移工具，需要结合tortoise异步orm框架使用。安装aerich
pip install aerich初始化配置，只需要使用一次
aerich init -t settings.TORTOISE_ORM # TORTOISE_ORM配置的位置)初始化完会在当前目录生成一个文件：pyproject.toml和一个文件夹：migrations
pyproject.toml：保存配置文件路径，低版本可能是aerich.ini
migrations：存放迁移文件
初始化数据库，一般情况下只用一次
此时数据库中就有相应的表格
如果TORTOISE_ORM配置文件中的models改了名，则执行这条命令时需要增加--app参数，来指定你修改的名字
更新模型并进行迁移
修改model类，重新生成迁移文件,比如添加一个字段
class Admin(Model):
...
xxx = fields.CharField(max_length=255)aerich migrate [--name] (标记修改操作) # aerich migrate --name add_column迁移文件名的格式为 {version_num}{datetime}{name|update}.json。 注意，此时sql并没有执行，数据库中admin表中没有xxx字段
重新执行迁移，写入数据库
aerich upgrade回到上一个版本
aerich downgrade查看历史迁移记录
aerich history7.系统接口开发 api/student.py
from fastapi.exceptions import HTTPException
from models import *
from pydantic import BaseModel
from typing import List, Union
from fastapi import APIRouter
api_student = APIRouter()
@api_student.get("/student")
async def getAllStudent():
students = await Student.all().values("name", "clas__name")
# students = await Student.filter(name__icontains='a').values("name", "clas__name")
# print("students", students)
# for i in students:
# print(i)
#
# rain = await Student.get(name='rain')
# print(rain, type(rain))
# print(rain.sno)
return students
class StudentModel(BaseModel):
name: str
pwd: str
sno: int
clas_id: Union[int, None] = None
courses: List[int] = []
@api_student.post("/student")
async def addStudent(stu: StudentModel):
# 添加数据库操作
# 方式1
# student = Student(name=stu.name, pwd=stu.pwd, sno=stu.sno, clas_id=stu.clas)
# await student.save()
# 方式2
student = await Student.create(name=stu.name, pwd=stu.pwd, sno=stu.sno, clas_id=stu.clas_id)
print(student, dir(student))
# 添加多对多关系记录
courses = await Course.filter(id__in=stu.courses)
print("courses", courses)
await student.courses.add(*courses)
print("student", student.courses)
return student
@api_student.put("/student/{student_id}")
async def update_student(student_id: int, student: StudentModel):
data = student.dict(exclude_unset=True)
courses = data.pop("courses")
print(data, courses)
await Student.filter(id=student_id).update(**data)
courses = await Course.filter(id__in=student.courses)
edit_student = await Student.get(id=student_id)
await edit_student.courses.clear()
await edit_student.courses.add(*courses)
return student
@api_student.delete("/student/{student_id}")
async def delete_student(student_id: int):
deleted_count = await Student.filter(id=student_id).delete() # 条件删除
if not deleted_count:
raise HTTPException(status_code=404, detail=f"Student {student_id} not found")
return {}