FastAPi入门
web服务启功
1 ——直接点击运行符号（不用重启就能更新代码）
2 -——输入 uvicorn main:app --reload（pythonweb的热启动，不用重启就能更新代码）
交互文档
路径后加/docs
路由（Java里的controller）
路径请求方法
路径参数
from fastapi import FastAPI,Path
再形参后加=Path（）
查询参数
导入
from fastapi import FastAPI,Query
QUery=（）
请求体
先导包
from pydantic import BaseModel，Field
一参见实体类（类注解导入Field）
=Filed（）
二定义请求体
class User(BaseModel): usernamr:str password:str @app.post("/user") async def create_user(user:User): return user响应格式
导入
这里导入的为html格式
from starlette.responses import HTMLResponse
创建接口
@app.get("/html",response_class=HTMLResponse) async def html(): return "<h1> 这是个html格式</h1>"自定义响应格式
要先基于BaseModel创建实体类
然后创建接口
异常处理
先导包
from fastapi import HTTPException
然后定义接口
用raise HTTPException()抛出问题
中间件（类似拦截器）
先进的最后执行代码 ， 后进的的 先执行代码
依赖注入
一导入Depends
from fastapi import FastAPI,Depends
2抽取充分的异步函数
3依赖注入
形参名=Depends（抽取充分的异步函数 ）
orm建表
1创建异步引擎
2定义数据类型 ： 基类（提取共用的属性如：创建时间和更改时间）和对应表的数据模型
3定义函数来调用异步引擎
引入的包 from datetime import datetime from fastapi import FastAPI from sqlalchemy import func, DateTime,String from sqlalchemy.ext.asyncio import create_async_engine from sqlalchemy.orm import DeclarativeBase,Mapped,mapped_column #,Mapped为声明类型 ，mapped_column为映射字段#创建异步引擎 FastAPI_url="mysql+aiomysql://root:123456@localhost:3306/fastapi_first" engine = create_async_engine(FastAPI_url, echo=True) #2定义数据类型 创建基类+表对应的数据模型 class Datatime: pass class Base(DeclarativeBase): create_time :Mapped[datetime] = mapped_column(DateTime,insert_default=func.now,default=func.now,comment="创建时间") update_time : Mapped[datetime] = mapped_column(DateTime,insert_default=func.now,default=func.now,comment="更新时间") # 定义数据模型 ：书籍id class book(Base): __tablename__ = "book" # 表名 id :Mapped[int] =mapped_column( primary_key=True,comment="id") name :Mapped[str] =mapped_column(String(255),comment="书名") author :Mapped[str] =mapped_column(String(50),comment="作者") #3定义函数建表——》fastAP调用函数建表 async def create_table(): # 获取异步请求引擎 async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) @app.on_event("startup") async def startup(): await create_table()orm的使用
# 定义会话工厂 sessionmaker = async_sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False ) # 依赖项 async def get_session(): async with sessionmaker() as session: try: yield session await session.commit() except: await session.rollback() raise finally: await session.close() @app.get("/book") async def get_book(session=Depends(get_session)): book_list = await session.execute(select(book)) return book_list.scalars().all()这段代码实现：FastAPI + 异步 SQLAlchemy 数据库操作包含：会话工厂、依赖注入、事务管理、查询数据
二、逐段知识点讲解
1️⃣ async_sessionmaker（异步会话工厂）
python
运行
sessionmaker = async_sessionmaker( bind=engine, # 绑定数据库引擎 class_=AsyncSession, # 使用异步会话 expire_on_commit=False # 提交后不过期 )知识点
- sessionmaker：不是会话本身，是制造会话的工厂
- AsyncSession：SQLAlchemy 异步会话，必须配合
await使用 - expire_on_commit=False
- 提交后，查询到的对象不会失效
- 可以继续使用，不会报错
- 作用：统一创建数据库会话，避免重复写代码
2️⃣ 依赖项 get_session () —— 数据库连接管理器
python
运行
async def get_session(): async with sessionmaker() as session: try: yield session await session.commit() except: await session.rollback() raise finally: await session.close()核心知识点
- async with
- 异步上下文管理器
- 自动创建 / 关闭会话
- yield session
- 生成器依赖
- 先给接口提供会话
- 请求结束后再执行后续代码（commit/rollback）
- 事务机制
commit()：成功就提交rollback()：出错就回滚，保证数据安全
- finally
- 无论成功失败，一定关闭会话
- 防止数据库连接泄漏
3️⃣ Depends 依赖注入
python
运行
async def get_book(session=Depends(get_session)):知识点
- Depends：FastAPI 依赖注入
- 作用：
- 自动调用
get_session() - 自动创建会话
- 自动提交 / 回滚
- 自动关闭
- 自动调用
- 让接口只关心业务，不关心数据库连接
4️⃣ 异步数据库查询
python
运行
book_list = await session.execute(select(book)) return book_list.scalars().all()知识点
- await
- 所有异步数据库操作必须加
- session.execute()
- 执行 SQL 语句的标准方法
- select(book)
- SQLAlchemy 2.0 新式查询语法
- 等价于
SELECT * FROM book
- scalars().all()
scalars()：只取模型对象，去掉元组包装all()：返回所有结果列表- 结果：
[Book对象, Book对象...
orm查询
@app.get("/book") async def get_book(db: AsyncSession = Depends(get_session)): # book_list = await db.execute(select(book))# 查询——》返回一个orm对象 # result=book_list.scalars().all()# 获取所有数据 # result = book_list.scalars().first() # 获取一条数据 books = await db.get(book, 2) return booksorm条件查询
模糊查询
示例代码 @app.get("/book/{id}") async def getbookByid(id:int,db:AsyncSession = Depends(get_session)): result=await db.execute(select(book).where(book.id==id)) books = result.scalar_one_or_none() return books 1. @app.get("/book/{id}") 作用：定义一个 GET 接口 /book/{id}：动态路由 {id}：表示可以传入任意数字，比如 /book/2、/book/3 效果：用户访问 /book/2 就会执行下面的函数 2. async def getbookByid(id:int, db:AsyncSession = Depends(get_session)): async def：异步函数，FastAPI 必须这么写 getbookByid：函数名（根据 ID 获取书籍） id:int：接收前端传来的 id 参数，并限定必须是数字 db:AsyncSession = Depends(get_session) 从依赖项获取数据库会话 db 就是你和数据库通信的 “通道” 3. result = await db.execute(select(book).where(book.id == id)) 这一行是核心查询语句 await：必须加！ 异步操作等待执行 db.execute(...)：执行 SQL 语句 select(book)：等价于 SELECT * FROM book .where(book.id == id)：条件查询 意思：只查 book 表里 id = 传入参数的那条数据 整体等价于 SQL： sql SELECT * FROM book WHERE id = ? 4. books = result.scalar_one_or_none() result：是数据库返回的原生结果集 .scalar_one_or_none() 取出一条结果 有数据 → 返回对象 没数据 → 返回 None 最安全、最常用的取值方法 5. return books 把查询到的书籍数据返回给前端 如果有数据 → 返回 JSON 没有 → 返回 null聚合查询
分页查询
示例 # 1. 定义 GET 接口地址：/book/select @app.get("/book/select") async def select_book( page=1, # 默认第 1 页 page_size=2, # 默认每页 2 条数据 db: AsyncSession = Depends(get_session) # 数据库会话 ): # 计算：从第几条开始查 # 第1页：(1-1)*2 = 0 → 从第0条开始 # 第2页：(2-1)*2 = 2 → 从第2条开始 skil = (page - 1) * page_size # 构建查询语句：分页查询 # select(book) → 查询 book 表 # offset(skil) → 跳过 skil 条数据 # limit(page_size) → 只取 page_size 条 stment = select(book).offset(skil).limit(page_size) # 2. 执行数据库查询 result = await db.execute(stment) # 3. 获取所有查询到的书籍列表 # scalars() → 提取 ORM 对象 # all() → 转成列表 books = result.scalars().all() # 4. 返回数据给前端 return {"book": books}新增
book在数据库层面是表，bookBase你可以把它理解为表单校验
class bookBase(BaseModel): id:int name:str author:str @app.post("/book/add") async def add_booK(book_data: bookBase ,db:AsyncSession = Depends(get_session)): book_obj = book(**book_data.__dict__)# 将数据模型转换为orm对象 db.add(book_obj)# 添加数据 await db.commit()# 提交会话 return book_data更新
查找——》重新赋值
@app.put("/book/update/{id}") async def update_book(id:int,book_data:bookBase,db:AsyncSession = Depends(get_session)): #查询数据 books =await db.get(book,id) if books is None: return {"message": "数据不存在"} #更新数据 books.name=book_data.name books.author=book_data.author await db.commit() return books删除数据
查找——》删除
@app.delete("/book/delete/{id}") async def delete_book(id:int,db:AsyncSession = Depends(get_session)): #查询数据 books =await db.get(book,id) if books is None: return {"message": "数据不存在"} #删除数据 await db.delete(books) await db.commit() return {"message": "删除成功"}redis引用
封装redis方法
新闻项目实战：
用户接口
用户注册接口
实现流程
问题：添加insert 和add（）的区别
方法： insert 是对 sql的insert方法的封装接近sql语句属于orm的低级语句
add是orm的高级语句需要有模型类
add是 ORM 的高层接口，用于将一个模型对象添加到当前的会话中进行管理。在 SQLAlchemy 中，它与Session绑定，将模型实例置于会话的“脏”状态，允许后续的事务提交来实际完成插入操作。
特点：
需要通过 ORM 的模型类创建一个对象实例，并将实例与会话绑定。
隐式地处理对象间关系（例如外键约束）和其他高级机制。
暂时将对象保存到会话中，只有当
commit或flush时，插入操作才会真正被发送到数据库。支持更丰富的功能，比如对象状态跟踪（
transient、pending、persistent等）。
适用场景：
需要利用 ORM 提供的高层功能，例如关系自动管理、级联保存等。
通常用于需要动态创建和插入单个或少量对象的情况。
SQLAlchemy 示例：
# 假设 User 是一个 ORM 模型 user = User(name="Alice", age=30) # 创建一个新的用户对象 session.add(user) # 将用户添加到会话 session.commit() # 提交事务，将对象数据插入数据库总结对比
特性 |
|
|
|---|---|---|
操作级别 | 偏底层，接近原生 SQL | 高级，基于 ORM 的对象管理 |
插入方式 | 批量或直接插入 SQL 语句 | 创建模型对象后添加到 ORM 会话 |
关系处理 | 不处理 | 自动处理对象关系 |
性能 | 性能较高（单纯插入） | 性能稍慢（带有对象跟踪和关系管理） |
应用场景 | 大规模插入或对简单数据的直接操控 | 对象化管理、动态插入及 ORM 全功能支持 |
问题：pyanctic作用
解决：做请求体，响应体，数据校验和类型转化的模型类
一、核心作用
专门用来接口请求体校验、响应体格式化、自动类型转换、字段规则约束，FastAPI 标配。
二、请求体（入参模型）
继承BaseModel定义字段，自动接收并校验前端传参。
python
运行
from pydantic import BaseModel class UserReq(BaseModel): name: str age: int email: str | None = None三、响应体（出参模型）
同样用BaseModel，规范返回给前端的字段、过滤多余字段、统一格式。
python
运行
class UserResp(BaseModel): id: int name: str # 只返回定义的字段，自动过滤多余属性四、数据转换
自动类型强转：字符串数字自动转
int/float时间字符串自动转
datetime字典自动解析成嵌套模型
模型可一键转字典 / JSON：
model.model_dump()→ 字典model.model_dump_json()→ JSON 字符串
五、数据约束（字段限制）
用Field做规则约束
数值：
gt大于、ge大于等于、lt小于、le小于等于字符串：
min_length/max_length长度限制、regex正则格式校验：
EmailStr邮箱、HttpUrl网址固定枚举：
Literal["男","女"]只能选指定值
六、自定义约束
装饰器@field_validator手写校验规则，不符合直接抛异常。
七、额外配置
model_config = {"extra": "forbid"}禁止传多余未知字段，防脏数据。
问题：什么时间用await
一、必须用 await 的 4 种场景
调用 async 定义的函数
python
运行
async def func(): ... await func() # 必须加操作数据库（异步驱动）
SQLAlchemy async
Databases
Motor (MongoDB)
python
运行
await db.execute(...) await db.fetch_all(...)发送 HTTP 请求（异步客户端）
httpx.AsyncClient
aiohttp
python
运行
await client.get("https://xxx")异步 I/O 操作文件、Redis、消息队列等异步库
python
运行
await redis.get("key")
问题：redis 使用
1，下载redis
2连接redis
同步普通连接
python
运行
import redis r = redis.Redis( host="localhost", port=6379, db=0, decode_responses=True # 自动字节转字符串，不用手动encode/decode )异步连接（FastAPI 常用）
bash
运行
pip install redis[asyncio]python
运行
from redis.asyncio import Redis redis_client = Redis(host="localhost", port=6379, db=0, decode_responses=True)3封装方法
调用方法
前面链接起的变量名 redis_client,方法名
1. 字符串 String（最简单）
存普通文本、验证码、token
存：
set设值，setex顺便设过期时间取：
get删：
delete查有没有：
exists
2. 哈希 Hash（专门存对象）
适合存用户、商品信息，不用拆成多个 key
单个字段存 / 取：
hset/hget一次性拿全部信息：
hgetall删除某个字段：
hdel
3. 列表 List（排队顺序用）
有序、可重复，适合消息队列、记录流水
左右加数据：
lpush、rpush左右弹出拿走数据：
lpop、rpop查全部：
lrange 0 -1
4. 集合 Set（自动去重）
无序、自动去重，适合签到、好友列表、黑名单
添加：
sadd查所有：
smembers删元素：
srem判断在不在集合里：
sismember
5. 有序集合 ZSet（做排行榜）
带分数自动排序，适合榜单、积分排名
加数据带分数：
zadd按排名查：
zrange删元素：
zrem
问题：pydantic 模型类调用
BaseModel
orm基本类调用 基类（存放一些通用字段）和模型类
DeclarativeBase
问题：scalars（）作用
scalars会给结果剥壳
scalars()：只拿查询的第一列字段，自动把外层列表、元组拆掉，直接拿到实体对象 / 单个值
一、基础作用
查出结果只取第一个字段
自动剥离
(对象,)这种元组包裹配合 ORM 查询，直接得到模型对象列表
不用自己循环拆元组，代码更简洁
二、常用链式方法 & 作用
1.result.scalars().all()
作用：查出全部数据，返回列表
返回：
[模型对象1, 模型对象2, ...]场景：查多条所有记录
2.result.scalars().first()
作用：只拿第一条数据，没有返回
None场景：查第一条、最新一条
3.result.scalars().one()
作用：必须有且只能有一条
多一条 / 少一条 直接报错
场景：确定唯一数据，比如按主键查
4.result.scalars().one_or_none()
作用：有一条就返回，没有返回
None，多条报错场景：不确定有没有，防止报错
5.result.scalars().limit(n)
作用：限制只取前 n 条
6.result.scalars().unique()
作用：去重，配合分组、联表查询去重
7. 直接遍历for obj in result.scalars()
作用：直接循环每一个模型对象，不用拆包
三、对比有无 scalars () 区别
不用 scalars ()
python
运行
res = db.execute(select(User)) all_data = res.all() # 结果：[(user1,), (user2,)] 带元组，还要拆用 scalars ()
python
运行
res = db.execute(select(User)).scalars().all() # 结果：[user1, user2] 直接是对象列表四、一句话口诀
scalars()：剥元组、只取首列、直拿对象.all()拿全部.first()拿第一条.one()必须唯一一条.one_or_none()有就拿，没有为空.unique()去重
问题orm应用
二、配置数据库连接 URL
不同数据库格式不同，填你的账号密码：
python
运行
# 同步URL # MySQL: mysql+pymysql://账号:密码@地址:端口/库名 DATABASE_URL = "mysql+pymysql://root:123456@localhost:3306/test_db" # SQLite: sqlite:///本地文件路径 # DATABASE_URL = "sqlite:///./test.db" # 异步URL（把驱动换成async版） ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/test_db"三、创建数据库引擎
连接数据库的核心入口：
python
运行
# 同步引擎 from sqlalchemy import create_engine # echo=True 会打印执行的SQL，调试用，生产关 engine = create_engine(DATABASE_URL, echo=True) # 异步引擎 from sqlalchemy.ext.asyncio import create_async_engine async_engine = create_async_engine(ASYNC_DATABASE_URL, echo=True)四、创建 ORM 基类
所有模型的父类，继承它才能被 ORM 识别：
python
运行
from sqlalchemy.orm import declarative_base Base = declarative_base()五、定义 ORM 模型（表结构）
类 = 表，属性 = 字段，对象 = 行：
python
运行
from sqlalchemy import Column, Integer, String # 这就是你的表 class User(Base): __tablename__ = "users" # 数据库里的表名 # 字段定义 id = Column(Integer, primary_key=True, index=True) # 主键 name = Column(String(50), index=True) # 字符串，最大50 age = Column(Integer)六、自动建表
把你定义的模型，同步到数据库，生成真实的表：
python
运行
# 同步版，直接跑 Base.metadata.create_all(bind=engine) # 异步版，要在异步函数里跑 async def init_db(): async with async_engine.begin() as conn: await conn.run_sync(Base.metadata.create_all)七、创建会话（操作数据库的句柄）
有了会话，才能真正增删改查：
python
运行
# 同步会话 from sqlalchemy.orm import sessionmaker SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # FastAPI 常用依赖写法，直接用 def get_db(): db = SessionLocal() try: yield db # 把会话给接口用 finally: db.close() # 用完自动关 # 异步会话 from sqlalchemy.ext.asyncio import async_sessionmaker AsyncSessionLocal = async_sessionmaker(autocommit=False, autoflush=False, bind=async_engine) # FastAPI 异步依赖 async def get_db(): async with AsyncSessionLocal() as session: yield session
