FASTAPI第二天-尧图网络科技

FastAPI入门以及代码进化

一、什么是ORM？

1.1ORM的概念以及优势

二、环境搭建和FastAPI集成

2.1 安装依赖
2.2 与MYSQL的使用
2.3项目结构搭建
2.4数据库配置文件

三、增删改查的实现

3.1查找
3.2增加
3.3修改
3.4删除

1.1.1

ORM全称为：Object-Relational Mapping（对象关系映射）、它的核心思想是用pychon类来代表数据库中的表

1.1.2

ORM 的优势：

面向对象：用 Python 代码替代 SQL 语句，更符合编程思维。
安全性：自动进行参数化查询，防止 SQL 注入攻击。
跨数据库：同一套代码可以无缝切换 SQLite、PostgreSQL、MySQL 等数据库。
关系管理：自动处理表与表之间的外键、多对多等关系。
可维护性：表结构集中定义在模型类中，修改和管理更方便。

二、环境搭建和FastAPI集成

2.1 安装依赖

要使用 Tortoise-ORM 与 FastAPI 集成，需要安装以下依赖：

# 安装 Tortoise-ORM 核心库pipinstalltortoise-orm# MySQL 异步驱动（推荐 asyncmy）pipinstallasyncmy# 或者使用 aiomysql# pip install aiomysql# 安装 Aerich 迁移工具（用于数据库迁移）pipinstallaerich# 安装 FastAPI 和 Uvicornpipinstallfastapi"uvicorn[standard]"

依赖说明：

tortoise-orm: 异步 ORM 核心库，提供模型定义、查询等 ORM 功能
asyncmy: 高性能 MySQL 异步驱动（推荐）
aiomysql: 另一个 MySQL 异步驱动（备选）
aerich: Tortoise-ORM 的数据库迁移工具
fastapi: 现代、快速的 Web 框架
uvicorn: ASGI 服务器，用于运行 FastAPI 应用

验证安装：
安装完成后，可以创建一个简单的 Python 文件验证依赖是否安装成功：

# test_install.pyimporttortoiseimportasyncmyimportaerichimportfastapiimportuvicornprint("Tortoise-ORM version:",tortoise.__version__)print("FastAPI version:",fastapi.__version__)print("Uvicorn version:",uvicorn.__version__)print("所有依赖安装成功！")

运行验证脚本：

python test_install.py

如果一切正常，将看到各个库的版本信息输出。

三、增删改查的实现

3.1 查找

常用方法包括：

await Model.all()：查询全部记录。
await Model.filter(...)：按条件过滤。
await Model.get_or_none(...)：按条件查询单条，查不到返回None，避免抛异常。
await Model.filter(...).count()：统计数量。
await Model.all().order_by("-priority")：按字段排序（-表示倒序）。

示例：获取所有任务并统计状态

@task_api.get("/all",summary="获取所有任务信息",description="获取所有任务信息")asyncdefgetAllTask():tasks1=awaitTasks.all().order_by("-priority")count1=awaitTasks.filter(status=1).count()count0=awaitTasks.filter(status=0).count()count2=awaitTasks.filter(status=2).count()return{"code":1,"message":"success","data":{"任务列表":tasks1,"总任务数量":len(tasks1),"待办数量":count0,"进行中数量":count1,"已完成数量":count2}}

这里的Tasks.all()会返回全部任务；order_by("-priority")表示按优先级从高到低排序；filter(status=0).count()则统计不同状态的任务数量。

示例：按 ID 查找单条任务

@task_api.get('/today_count',summary="今日到期任务数量",description="统计今日到期且未完成的任务数量")asyncdeftoday_count():today=datetime.now().replace(hour=0,minute=0,second=0,microsecond=0)tomorrow=today+timedelta(days=1)count=awaitTasks.filter(due_date__gte=today,due_date__lt=tomorrow,status__in=[0,1]).count()return{"code":1,"message":"success","data":count}

filter(due_date__gte=today, due_date__lt=tomorrow, status__in=[0,1])展示了条件查询：

__gte：大于等于（greater than or equal）。
__lt：小于（less than）。
__in：字段值在列表中。

示例：获取所有用户

@user_api.get("/all",summary="获取所有用户信息",description="获取所有用户信息")asyncdefgetAllUser():users1=awaitUser.all()return{"code":1,"message":"success","data":users1}

3.2 增加

新增数据有两种常见方式：

await Model.create(**kwargs)：直接创建并保存到数据库，返回模型实例。
model = Model(**kwargs); await model.save()：先实例化，再手动保存。

示例：创建任务

@task_api.post('/add',summary="添加任务",description="添加任务")asyncdefadd(task:TaskAdd):data=task.model_dump()ifhasattr(task,"model_dump")elsedict(task)data["user_id"]=data.pop("user")task1=awaitTasks.create(**data)return{"code":1,"message":"添加成功","data":task1}

这里TaskAdd是 Pydantic 请求模型。由于Tasks表中的外键字段名是user_id，但前端传入的是user，所以通过data["user_id"] = data.pop("user")做了一个字段映射。最后调用Tasks.create(**data)完成插入。

示例：创建用户

@user_api.post("/add",summary="添加用户",description="添加用户")asyncdefgetadd(user:UseraddRequest):user2=User(**dict(user))awaituser2.save()return{"code":1,"message":"添加成功","data":user2}

这里先用User(**dict(user))实例化模型对象，再调用await user2.save()保存到数据库。两种方式效果相同，选择哪种主要看个人习惯。

3.3 修改

修改流程通常是：

先用get_or_none查出目标记录。
判断记录是否存在，不存在则返回提示。
修改对应字段。
调用await model.save()保存。

示例：更新任务信息

@task_api.put('/update/{id}',summary="更新任务信息",description="更新任务信息")asyncdefedit(task:TaskEdit,id:int):task3=awaitTasks.get_or_none(id=id)iftask3isNone:return{"code":0,"message":"任务不存在","data":None}task3.user=task.user task3.title=task.title task3.description=task.description task3.status=task.status task3.priority=task.priority task3.due_date=task.due_date task3.complete_date=task.complete_dateawaittask3.save()return{"code":1,"message":"更新成功","data":task3}

示例：更新任务状态

@task_api.put('/{id}/status',summary="切换任务状态",description="切换到已完成时自动记录完成时间，切回其他状态时清空完成时间")asyncdefupdate_status(id:int,status:int):task=awaitTasks.get_or_none(id=id)iftaskisNone:return{"code":0,"message":"任务不存在","data":None}task.status=statusifstatus==2:task.complete_date=datetime.now()else:task.complete_date=Noneawaittask.save()return{"code":1,"message":"状态切换成功","data":task}

update_status里除了修改状态，还根据业务规则自动处理complete_date：切换到已完成时记录当前时间，切回其他状态时清空完成时间。

示例：更新用户信息

@user_api.put("/update/{id}",summary="更新用户信息",description="更新用户信息")asyncdefedit(user:UserEditRequest,id:int):user3=awaitUser.get_or_none(id=id)ifuser3isNone:return{"code":0,"message":"用户不存在","data":None}user3.username=user.username user3.email=user.email user3.phone=user.phone user3.is_active=user.is_activeawaituser3.save()return{"code":1,"message":"更新成功","data":user3}

3.4 删除

删除流程与修改类似：先查到对象，再调用await model.delete()。

示例：删除任务

@task_api.delete('/delete/{id}',summary="删除任务",description="删除任务")asyncdefdelete(id:int):task4=awaitTasks.get_or_none(id=id)awaittask4.delete()return{"code":1,"message":"删除成功","data":None}

示例：删除用户

@user_api.delete("/delete/{id}",summary="删除用户",description="删除用户")asyncdefdelete(id:int):user4=awaitUser.get_or_none(id=id)awaituser4.delete()return{"code":1,"message":"删除成功","data":None}