基于FastAPI的AI智能体Web系统构建（三）

# ORM基础配置
from tortoise.contrib.fastapi import register_tortoise
from typing import Dict
from fastapi import FastAPI

app = FastAPI()


# Tortoise-ORM 配置
TORTOISE_ORM: Dict = {
    # connections：定义数据库连接字符串
    "connections": {
        # 开发环境使用 SQLite（基于文件，无需服务器）
        # "default": "sqlite://db.sqlite3",
        # 生产环境示例：PostgreSQL
        # "default": "postgres://user:password@localhost:5432/dbname",
        # 生产环境示例：MySQL
        "default": "mysql://root:123456@192.168.31.152:3306/fastapi_db",
    },
    # 定义应用模块，models 列表包含模型文件路径和Aerich的迁移模型
    "apps": {
        "models": {
            "models": ["model19", "aerich.models"],  # 模型模块和 Aerich 迁移模型
            "default_connection": "default", #这里的default指的就是connections中的default
        }
    },
    # 连接池配置（推荐）
    "use_tz": False,  # 是否使用时区
    "timezone": "UTC",  # 默认时区
    "db_pool": {
        "max_size": 10,  # 最大连接数
        "min_size": 1,   # 最小连接数
        "idle_timeout": 30  # 空闲连接超时（秒）
    }
}

# 注册 Tortoise-ORM 到 FastAPI 应用
register_tortoise(app,
                  config=TORTOISE_ORM,
                  generate_schemas=True,  # 开发环境自动生成表结构
                  add_exception_handlers=True  # 添加默认异常处理
                  )


if __name__ == '__main__':
    import uvicorn
    uvicorn.run('main19:app', host='127.0.0.1', port=8000, reload=True)

三.Aerich 迁移工具的使用

1.Aerich命令

Aerich 是 Tortoise-ORM 的数据库迁移工具，用于管理数据库结构的变更。

Aerich 初始化

在项目根目录运行以下命令：

aerich init -t main.TORTOISE_ORM

这将生成：

pyproject.toml ： Aerich 配置文件，指定迁移配置。

migrations/ ：迁移文件目录，存放生成的 .sql 文件。

aerich init-db (根据类来创建表)

生成和应用迁移

生成迁移文件：当模型发生变更时，运行以下命令生成迁移文件（生成SQL）：

aerich migrate --name "info"

应用迁移：运行以下命令将迁移应用到数据库(执行SQL)：

aerich upgrade

验证迁移：检查迁移历史

aerich history

回滚迁移：回退到指定版本

aerich downgrade

2.Aerich迁移

先安装database client插件

填写好信息，点连接

成功

点加号

在新连接这里点加号，创建数据库，点击Run

新建数据库成功

创建一个类文件model19.py

from tortoise.models import Model
from tortoise.fields import CharField, DatetimeField, BooleanField


class User(Model):
    id = CharField(max_length=36, pk=True)  # 主键，UUID 字符串
    username = CharField(max_length=50, unique=True)  # 用户名，唯一
    email = CharField(max_length=255, unique=True)  # 邮箱，唯一
    is_active = BooleanField(default=True)  # 是否激活
    email666 = CharField(max_length=255, unique=True)  # 邮箱，唯一

    class Meta:
        table = "users"  # 自定义表名
        ordering = ["-created_at"]  # 默认按创建时间降序排序

    def __str__(self):
        return self.username


'''
# 根据main19.py文件中的User模型去生成迁移文件,因为此文件里的TORTOISE_ORM变量相当于配置文件
# 这两个init初始化命令一般在项目周期里只执行一次
aerich init -t main19.TORTOISE_ORM   # 初始化
aerich init-db  # 根据写的类来创建表

# 这两个迁移命令一般在项目周期一般执行很多次，执行迁移(migrate)后都需要同步迁移（upgrade）到数据库
aerich migrate --name "注释"    # 迁移，即写好了图纸（生成SQL）
aerich upgrade  # 同步迁移到数据库，按照图纸施工（执行 SQL）
'''

配置和注册文件main19.py

# ORM基础配置
from tortoise.contrib.fastapi import register_tortoise
from typing import Dict
from fastapi import FastAPI

app = FastAPI()


# Tortoise-ORM 配置
TORTOISE_ORM: Dict = {
    # connections：定义数据库连接字符串
    "connections": {
        # 开发环境使用 SQLite（基于文件，无需服务器）
        # "default": "sqlite://db.sqlite3",
        # 生产环境示例：PostgreSQL
        # "default": "postgres://user:password@localhost:5432/dbname",
        # 生产环境示例：MySQL
        "default": "mysql://root:root@127.0.0.1:3306/fastapi_db",
    },
    # 定义应用模块，models 列表包含模型文件路径和Aerich的迁移模型
    "apps": {
        "models": {
            # 根据model19.py文件中的User模型去生成迁移文件
            "models": ["model19", "aerich.models"],  # 模型模块和 Aerich 迁移模型
            "default_connection": "default", #这里的default指的就是connections中的default
        }
    },
    # 连接池配置（推荐）
    "use_tz": False,  # 是否使用时区
    "timezone": "UTC",  # 默认时区
    "db_pool": {
        "max_size": 10,  # 最大连接数
        "min_size": 1,   # 最小连接数
        "idle_timeout": 30  # 空闲连接超时（秒）
    }
}

# 注册 Tortoise-ORM 到 FastAPI 应用
register_tortoise(app,
                  config=TORTOISE_ORM,
                  generate_schemas=True,  # 开发环境自动生成表结构
                  add_exception_handlers=True  # 添加默认异常处理
                  )


if __name__ == '__main__':
    import uvicorn
    uvicorn.run('main19:app', host='127.0.0.1', port=8000, reload=True)

依次执行那四个命令即可，model19.py文件里注释写的非常明白。

后续如果需要修改表结构，直接修改代码即可，比如把User类里的某个成员变量删了，然后执行迁移和同步，数据库就随之改变了，确实好用啊，太方便了。

aerich migrate --name "注释"    # 迁移

aerich upgrade  # 同步迁移到数据库

四.Tortoise-ORM 的模型定义

Tortoise-ORM 的模型通过继承 tortoise.models.Model 类定义，每个模型对应数据库中的一张表。

字段定义使用 Tortoise-ORM 提供的字段类型，结合字段参数来指定约束和行为。

继承 Model ：所有模型必须继承 tortoise.models.Model 。

字段定义 ：每个字段对应数据库表中的一列，使用 Tortoise-ORM 的字段类（如 CharField 、

BooleanField ）。

Meta 类：用于定义模型的元数据，如表名、排序规则等。

核心字段类型详解

Tortoise-ORM 提供了多种字段类型，用于定义数据库列的类型和行为。以下是常用的字段类型及其典型用途：

字段参数详解

字段支持多种参数，用于定义约束、默认值和其他行为。以下是常用参数：

max_length ：字符串字段的最大长度（ CharField 必填）

null ：是否允许字段为空（ null=True 表示数据库允许 NULL ）

default ：字段默认值（如 default=0 、 default=True ）

unique ：是否唯一（ unique=True 确保字段值在表中唯一）

index ：是否创建索引（ index=True 提高查询性能）

description ：字段描述（用于文档或数据库注释）

pk ：是否为主键（ pk=True 表示该字段是主键）

validators ：自定义验证函数

模型元数据（ Meta 类）

Meta 类用于定义模型的元数据，控制表结构和查询行为。常用属性包括：

table ：自定义数据库表名（如 table="users" ）

unique_together ：定义联合唯一约束（如 unique_together=[("field1", "field2")] ）

indexes ：定义索引（如 indexes=[("field1", "field2")] ）

ordering ：默认排序规则（如 ordering=["-created_at", "username"] ）

from tortoise.models import Model
from tortoise.fields import CharField, TextField, IntField, BooleanField, DatetimeField


class User(Model):
    id = CharField(max_length=36, pk=True)
    name = CharField(max_length=64, unique=True)
    email = CharField(max_length=255)
    is_active = BooleanField(default=True)
    age = IntField(default=0)
    # 创建时间
    created_at = DatetimeField(auto_now_add=True)

    class Meta:
        table = 't_user'  # table/view
        unique_together = ('name', 'email')  # 联合唯一
        ordering = ['-created_at']  # -降序 默认升序
        # 创建索引
        indexes = [
            'email',
        ]


'''
aerich init -t main20.TORTOISE_ORM   # 初始化
aerich init-db  # 创建迁移脚本

aerich migrate --name "注释"    # 迁移
aerich upgrade  # 同步迁移
'''

五.Tortoise-ORM CRUD增删改查

model21.py 学生类

from tortoise import fields, models


class Student(models.Model):
    """
    学生模型，包含基本信息。
    用于演示单表的增删改查操作。
    """
    id = fields.IntField(pk=True, description="学生ID，主键")
    name = fields.CharField(max_length=50, description="学生姓名")
    age = fields.IntField(null=True, description="学生年龄，可为空")
    email = fields.CharField(max_length=100, unique=True,
                             null=True, description="学生邮箱，唯一")

    class Meta:
        table = "students"

    def __str__(self):
        return f"Student: {self.name}, Age: {self.age}, Email: {self.email}"

main21.py 接口方式实现CRUD

# ORM基础配置
from crud21 import create_student
from tortoise.contrib.fastapi import register_tortoise
from typing import Dict
from fastapi import FastAPI

app = FastAPI()


# Tortoise-ORM 配置
TORTOISE_ORM: Dict = {
    "connections": {
        # 开发环境使用 SQLite（基于文件，无需服务器）
        # "default": "sqlite://db.sqlite3",
        # 生产环境示例：PostgreSQL
        # "default": "postgres://user:password@localhost:5432/dbname",
        # 生产环境示例：MySQL
        "default": "mysql://root:123456@192.168.31.152:3306/fastapi_db3",
    },
    "apps": {
        "models": {
            "models": ["model21", "aerich.models"],  # 模型模块和 Aerich 迁移模型
            "default_connection": "default",
        }
    },
    # 连接池配置（推荐）
    "use_tz": False,  # 是否使用时区
    "timezone": "UTC",  # 默认时区
    "db_pool": {
        "max_size": 10,  # 最大连接数
        "min_size": 1,   # 最小连接数
        "idle_timeout": 30  # 空闲连接超时（秒）
    }
}

register_tortoise(app,
                  config=TORTOISE_ORM,
                  generate_schemas=True,  # 开发环境自动生成表结构
                  add_exception_handlers=True  # 添加默认异常处理
                  )

#创建对象时自动在数据库中插入了
@app.get('/create')
async def create():
    stu = await create_student('孙权', 18, 'sunquan@163.com')
    return stu

# 生成一个传递参数创建学生数据的路由


@app.post('/create2')
async def create2(name: str, age: int = None, email: str = None):
    stu = await create_student(name, age, email)
    return stu

if __name__ == '__main__':
    import uvicorn
    uvicorn.run('main21:app', host='127.0.0.1', port=8000, reload=True)

crud21.py 脚本方式实现CRUD

from model21 import Student
from tortoise import Tortoise, run_async


async def create_student(name: str, age: int = None, email: str = None) -> Student:
    try:
        stu = await Student.create(name=name, age=age, email=email)
        return stu
    except Exception as e:
        print("创建失败！")


async def update_student(stu_id: int, name: str = None, age: int = None, email: str = None) -> Student:
    stu = await Student.get(id=stu_id)
    # 判断用户传递了哪个数据，如果没有传递，就不更新
    if name:
        stu.name = name
    if age is not None:
        stu.age = age
    if email is not None:
        stu.email = email
    # 保存
    await stu.save()
    return stu


async def delete_student(stu_id: int) -> None:
    stu = await Student.get(id=stu_id)
    await stu.delete()

# 单条数据


async def get_student(stu_id: int) -> Student:
    stu = await Student.get(id=stu_id)
    return stu
# 多条数据


async def get_students(name: str) -> list[Student]:
    '''
    完成匹配
    '''
    stus = await Student.filter(name=name)
    return stus


async def get_students2(name: str) -> list[Student]:
    '''
    模糊查询
    '''
    stus = await Student.filter(name__contains=name)
    return stus


async def get_students3() -> list[Student]:
    '''
    获取所有数据
    '''
    stus = await Student.all()
    return stus

# 编写查询名字为曹操的年龄等于20的记录


async def get_students4() -> list[Student]:
    stus = await Student.filter(name="曹操", age=20)
    return stus

    # 1. 直接测试脚本
    # 2. 直接应用到接口


async def init():
    await Tortoise.init(
        db_url="mysql://root:123456@192.168.31.152:3306/fastapi_db3",
        modules={"models": ["model21"]}
    )
    await Tortoise.generate_schemas()


async def main():
    await init()
    # stu1 = await create_student(name="曹操")
    # print(f"创建成功 ID:{stu1.id}  Name:{stu1.name} Email:{stu1.email}")
    # stu2 = await create_student(name="刘备", age=30)
    # print(f"创建成功 ID:{stu2.id}  Name:{stu2.name} Email:{stu2.email}")
    # stu3 = await create_student(name="张飞", age=30, email="zf@163.com")
    # print(f"创建成功 ID:{stu3.id}  Name:{stu3.name} Email:{stu3.email}")
    # stu4 = await create_student(name="关羽", age=30, email="zf@163.com")
    # print(f"创建成功 ID:{stu4.id}  Name:{stu4.name} Email:{stu4.email}")

    # await update_student(stu_id=3, age=36)
    # await update_student(stu_id=1, email="cc@qq.com", age=40)
    # await delete_student(stu_id=9)

    stu = await get_student(3)
    print(stu)
    stus1 = await get_students("刘备")
    print(stus1)
    stus2 = await get_students2("曹")
    print(stus2)
    stus3 = await get_students3()
    print(stus3)

if __name__ == '__main__':
    run_async(main())

六.Tortoise-ORM 的关联关系建立

我们将模拟一个学校管理系统，包含以下实体：

学生（ Student ）：每个学生有唯一的个人信息档案（ 1 对 1 ）。

成绩（ Grade ）：一个学生可以有多份成绩记录（ 1 对多）。

课程（ Course ）：学生和课程之间是多对多关系（通过成绩表关联）。

1.一对一

# 关联关系建立1对1
from tortoise import fields, models, Tortoise, run_async


class Student(models.Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=50)
    # 参数：on_delete=fields.CASCADE 表示当学生被删除时，其个人信息StudentProfile也会被删除
    # 参数：related_name="student" 表示在StudentProfile模型中，有一个字段指向Student模型，用于反向查询
    profile = fields.OneToOneField(
        "models.StudentProfile", on_delete=fields.CASCADE, related_name="student")


'''
id name   profile_id
1  张三     1
2  李四     2

id  address  phone      
1    上海     123456        
2    北京     789456       
'''


class StudentProfile(models.Model):
    id = fields.IntField(pk=True)
    address = fields.CharField(max_length=100)
    phone = fields.CharField(max_length=20)


async def init():
    await Tortoise.init(
        db_url="mysql://root:123456@192.168.31.152:3306/fastapi_db4",
        modules={"models": ["model22"]}
    )
    await Tortoise.generate_schemas()

if __name__ == "__main__":
    run_async(init())

2.一对多

# 关联关系建立1对多
from tortoise import fields, models, Tortoise, run_async


class Student(models.Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=50)
    profile = fields.OneToOneField(
        "models.StudentProfile", on_delete=fields.CASCADE, related_name="student")


'''
学生表
id name   profile_id  score_ids            score_id1   score_id2 ...
1  张三     1           1-2-3-5-8-6-7         1          2
2  李四     2

成绩表
id score stu_id
1  90     1
2  80     1
3  70     1
4  60     2
5  50     1
'''


class StudentProfile(models.Model):
    id = fields.IntField(pk=True)
    address = fields.CharField(max_length=100)
    phone = fields.CharField(max_length=20)

#一对多，在多的一方定义外键
class Grade(models.Model):
    id = fields.IntField(pk=True)
    score = fields.FloatField()
    student = fields.ForeignKeyField(
        "models.Student", related_name="grades", on_delete=fields.CASCADE)


async def init():
    await Tortoise.init(
        db_url="mysql://root:123456@192.168.31.152:3306/fastapi_db4",
        modules={"models": ["model23"]}
    )
    await Tortoise.generate_schemas()

if __name__ == "__main__":
    run_async(init())

3.多对多

model24.py

# 关联关系建立多对多
from tortoise import fields, models, Tortoise, run_async


'''
学生表
id name   profile_id     
1  张三     1             
2  李四     2

成绩表
id score stu_id
1  90     1
2  80     1
3  70     1
4  60     2
5  50     1

课程表
id name       
1  python     
2  java
3  c++

学生课程表
stu_id  course_id
1        1
1        2
2        3
'''


class Student(models.Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=50)
    profile = fields.OneToOneField(
        "models.StudentProfile", on_delete=fields.CASCADE, related_name="student", null=True)


class StudentProfile(models.Model):
    id = fields.IntField(pk=True)
    address = fields.CharField(max_length=100)
    phone = fields.CharField(max_length=20)


class Grade(models.Model):
    id = fields.IntField(pk=True)
    score = fields.FloatField()
    student = fields.ForeignKeyField(
        "models.Student", related_name="grades", on_delete=fields.CASCADE)


class Course(models.Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=50)
    # students = fields.ManyToManyField(
    #     "models.Student", related_name="courses", through="studentcourse")


class StudentCourse(models.Model):
    students = fields.ForeignKeyField("models.Student", related_name="courses")
    courses = fields.ForeignKeyField("models.Course", related_name="students")

    class Meta:
        unique_together = ("students", "courses")


async def init():
    await Tortoise.init(
        db_url="mysql://root:root@127.0.0.1:3306/fastapi_db4",
        modules={"models": ["model24"]}
    )
    await Tortoise.generate_schemas()

if __name__ == "__main__":
    run_async(init())

七.Tortoise-ORM 的关联关系数据操作

基于关联关系CRUD

# 关联关系的增删改查询
from tortoise import Tortoise, run_async
from model24 import Student, Course, StudentCourse, Grade, StudentProfile


async def init():
    await Tortoise.init(
        db_url="mysql://root:123456@192.168.31.152:3306/fastapi_db4",
        modules={"models": ["model24"]}
    )


async def create_data():
    await init()

    # 1对1
    # 学生对象的id会自动生成，profile对象是关联用的，所以不用必须赋初值
    # 下面两句create语句会落实到数据库
    # stu1 = await Student.create(name="董卓")
    # pro1 = await StudentProfile.create(address="上海", phone="123456")
    # 这句代码只是在内存里赋值，并不会落实到数据库，所以需要save操作
    # stu1.profile = pro1
    # 保存学生对象，会自动保存关联的profile对象，并落实到数据库
    # await stu1.save()

    # pro2 = await StudentProfile.create(address="北京", phone="123456")
    # stu2 = await Student.create(name="刘备", profile=pro2)

    # 1对多
    stu3 = await Student.create(name="孙权")
    await Grade.create(score=100, student=stu3)
    await Grade.create(score=90, student=stu3)
    await Grade.create(score=80, student=stu3)

    # 多对多
    stu4 = await Student.create(name="关羽")
    stu5 = await Student.create(name="张飞")
    cou1 = await Course.create(name="Python")
    cou2 = await Course.create(name="C++")
    await StudentCourse.create(students=stu4, courses=cou1)
    await StudentCourse.create(students=stu4, courses=cou2)
    await StudentCourse.create(students=stu5, courses=cou1)
    await StudentCourse.create(students=stu5, courses=cou2)


async def update_data():
    await init()
    # 1对1
    # 先从数据库取出
    stu = await Student.get(id=1)
    pro = await StudentProfile.create(address="上海", phone="12345678901")
    stu.profile = pro
    # 存回数据库
    await stu.save()

    # 1对多
    # 方式一
    stu2 = await Student.get(id=1)
    grade1 = await Grade.get(id=1)
    grade1.student = stu2
    await grade1.save()
    # 方式二
    await Grade.filter(id=6).update(student=stu2)

async def query_data():
    await init()
    stu = await Student.get(id=1)
    print(stu.name)
    # 关联对象profile需要异步获取  方式一
    pro = await stu.profile
    print(pro.address)
    print(pro.phone)
    # 关联对象profile需要异步获取  方式二
    stu2 = await Student.get(id=1).prefetch_related('profile')  # 获取关联数据
    print(stu2.profile.address)
    print(stu2.profile.phone)

    #all()加不加都行，一般不影响查询结果
    stu3 = await Student.get(id=9).prefetch_related('grades').all()
    for grade in stu3.grades:
        print(grade.score)


async def delete_data():
    await init()

    # grade = await Grade.get(id=9)
    # await grade.delete()

    #级联删除
    stu = await Student.get(id=9)
    await stu.delete()
    
if __name__ == '__main__':
    # run_async(create_data())
    # run_async(update_data())
    # run_async(query_data())
    run_async(delete_data())