FastAPI入门以及代码进化一、什么是ORM1.1ORM的概念以及优势二、环境搭建和FastAPI集成2.1 安装依赖2.2 与MYSQL的使用2.3项目结构搭建2.4数据库配置文件三、增删改查的实现3.1查找3.2增加3.3修改3.4删除1.1.1ORM全称为Object-Relational Mapping对象关系映射、它的核心思想是用pychon类来代表数据库中的表1.1.2ORM 的优势面向对象用 Python 代码替代 SQL 语句更符合编程思维。安全性自动进行参数化查询防止 SQL 注入攻击。跨数据库同一套代码可以无缝切换 SQLite、PostgreSQL、MySQL 等数据库。关系管理自动处理表与表之间的外键、多对多等关系。可维护性表结构集中定义在模型类中修改和管理更方便。二、环境搭建和FastAPI集成2.1 安装依赖要使用 Tortoise-ORM 与 FastAPI 集成需要安装以下依赖# 安装 Tortoise-ORM 核心库pipinstalltortoise-orm# MySQL 异步驱动推荐 asyncmypipinstallasyncmy# 或者使用 aiomysql# pip install aiomysql# 安装 Aerich 迁移工具用于数据库迁移pipinstallaerich# 安装 FastAPI 和 Uvicornpipinstallfastapiuvicorn[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():tasks1awaitTasks.all().order_by(-priority)count1awaitTasks.filter(status1).count()count0awaitTasks.filter(status0).count()count2awaitTasks.filter(status2).count()return{code:1,message:success,data:{任务列表:tasks1,总任务数量:len(tasks1),待办数量:count0,进行中数量:count1,已完成数量:count2}}这里的Tasks.all()会返回全部任务order_by(-priority)表示按优先级从高到低排序filter(status0).count()则统计不同状态的任务数量。示例按 ID 查找单条任务task_api.get(/today_count,summary今日到期任务数量,description统计今日到期且未完成的任务数量)asyncdeftoday_count():todaydatetime.now().replace(hour0,minute0,second0,microsecond0)tomorrowtodaytimedelta(days1)countawaitTasks.filter(due_date__gtetoday,due_date__lttomorrow,status__in[0,1]).count()return{code:1,message:success,data:count}filter(due_date__gtetoday, due_date__lttomorrow, status__in[0,1])展示了条件查询__gte大于等于greater than or equal。__lt小于less than。__in字段值在列表中。示例获取所有用户user_api.get(/all,summary获取所有用户信息,description获取所有用户信息)asyncdefgetAllUser():users1awaitUser.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):datatask.model_dump()ifhasattr(task,model_dump)elsedict(task)data[user_id]data.pop(user)task1awaitTasks.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):user2User(**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):task3awaitTasks.get_or_none(idid)iftask3isNone:return{code:0,message:任务不存在,data:None}task3.usertask.user task3.titletask.title task3.descriptiontask.description task3.statustask.status task3.prioritytask.priority task3.due_datetask.due_date task3.complete_datetask.complete_dateawaittask3.save()return{code:1,message:更新成功,data:task3}示例更新任务状态task_api.put(/{id}/status,summary切换任务状态,description切换到已完成时自动记录完成时间切回其他状态时清空完成时间)asyncdefupdate_status(id:int,status:int):taskawaitTasks.get_or_none(idid)iftaskisNone:return{code:0,message:任务不存在,data:None}task.statusstatusifstatus2:task.complete_datedatetime.now()else:task.complete_dateNoneawaittask.save()return{code:1,message:状态切换成功,data:task}update_status里除了修改状态还根据业务规则自动处理complete_date切换到已完成时记录当前时间切回其他状态时清空完成时间。示例更新用户信息user_api.put(/update/{id},summary更新用户信息,description更新用户信息)asyncdefedit(user:UserEditRequest,id:int):user3awaitUser.get_or_none(idid)ifuser3isNone:return{code:0,message:用户不存在,data:None}user3.usernameuser.username user3.emailuser.email user3.phoneuser.phone user3.is_activeuser.is_activeawaituser3.save()return{code:1,message:更新成功,data:user3}3.4 删除删除流程与修改类似先查到对象再调用await model.delete()。示例删除任务task_api.delete(/delete/{id},summary删除任务,description删除任务)asyncdefdelete(id:int):task4awaitTasks.get_or_none(idid)awaittask4.delete()return{code:1,message:删除成功,data:None}示例删除用户user_api.delete(/delete/{id},summary删除用户,description删除用户)asyncdefdelete(id:int):user4awaitUser.get_or_none(idid)awaituser4.delete()return{code:1,message:删除成功,data:None}小结通过上面的代码可以看到Tortoise-ORM 下的 CRUD 操作非常接近自然语言查all()、filter()、get_or_none()、count()增create()或save()改查出来 → 改字段 →save()删查出来 →delete()所有操作都别忘了加await因为 Tortoise-ORM 是原生异步的。另外真实的业务代码里通常会做空值判断if task is None和字段映射如user_id与user以保证接口健壮性。