FastAPI基础: 接口+ORM 使用指南-编程阁

FastAPI写接口

使用指南1

# 开发阶段pipinstall"fastapi[standard]"# 包含核心fastapi,uvicoorn,pydantic等常用依赖# 生产阶段#pip install fastapi, "uvicorn[standard]"# 用于启动服务端uvicorn main:app--reload

fromfastapiimportFastAPI app=FastAPI()# 创建 FastAPI 对象@app.get('/')# 装饰器，指定请求路径defroot():return{'message':'Hello World'}

问题1

为什么选择 FastAPI？
- 开发效率高
- 异步性能高
- 自动生成文档
请求路径如何与函数关联？
- 哪个函数写在装饰器 @app.get(路径) 或 @app.post(路径) 的下面, 这个函数就跟这个路径绑定
函数一定要写成async def ？
- 操作 MySQL、普通逻辑 -> def
- 做异步请求、异步数据库 -> async def
如何传参?
- 路径参数：适合指向唯一确定的资源
- 查询参数：适合分页，排序，筛选资源
- 请求体参数：适合创建, 更新资源,携带大量的数据(统一格式JSON)
返回响应是什么类型?
- 一般为python对象(dict), fastapi会自动转换成JSON格式
- from fastapi.responses import HTMLResponse, FileResponse
- 其它类型，需要自己转换如 -> return HTMLResponse(),return FileResponse()
- 其中响应文件类型，包含 -> 图片, 视频, 音频, excel, word, pdf
- 还有一种方法设置返回响应 -> @app.get(路径, response_class=HTMLResponse)
异常响应处理: 200/201/400/404/500 状态码的使用场景
- 通过导入 from fastapi import HTTPException
- 抛出异常 -> raise HTTPException(status_code=404, detail=‘找不到资源’)
- 查询 / 获取 / 修改 / 删除成功 → 200
- 新增 / 创建成功 → 201
- 参数错、格式错、校验失败 → 400
- 数据找不到、接口不存在 → 404
- 代码报错、数据库崩了 → 500
中间件函数: 拦截请求，对请求进行预处理，对响应进行后处理

@app.middleware('http')asyncdefmiddleware(request,call_next):print('请求开始')response=awaitcall_next(request)print('请求结束')returnresponse# 多个中间件的执行顺序: 按代码自下而上

FastAPI连接MySQL

使用指南2

pipinstallsqlalchemy, pymysql

fromfastapiimportFastAPI,Dependsfromsqlalchemyimportcreate_engine,Integer,String,DateTime,func,selectfromsqlalchemy.ormimportDeclarativeBase,sessionmaker,Mapped,mapped_column app=FastAPI()# 1.创建连接引擎-锁定数据库DB_URL="mysql+pymysql://root:密码@localhost:3306/数据库"engine=create_engine(DB_URL)# 2.定义自己地基类(先继承官方DeclarativeBase)classBase(DeclarativeBase):__abstract__=True# 表示抽象基类,不会生成表create_time:Mapped[datetime]=mapped_column(DateTime,server_default=func.now(),nullable=False)update_time:Mapped[datetime]=mapped_column(DateTime,server_default=func.now(),onupdate=func.now(),nullable=False)# 3.定义表模型类(继承地基类Base)classUser(Base):__tablename__='users'id:Mapped[int]=mapped_column(Integer,primary_key=True,autoincrement=True)username:Mapped[str]=mapped_column(String(32),nullable=False,unique=True)password:Mapped[str]=mapped_column(String(64),nullable=False)# 4.同步操作,自动建表Base.metadata.create_all(engine)## 操作数据库(增删改查)# 1.创建会话工厂SessionLocal=sessionmaker(bind=engine)# 2.依赖项defget_db():# 创建会话连接-真正连接到数据库db=SessionLocal()try:yielddb db.commit()exceptException:db.rollback()raisefinally:db.close()# 3. 查询所有用户@app.get("/users")defget_all_users(db=Depends(get_db)):# select(表)users=db.scalars(select(User)).all()return{"users":users}# 4. 根据ID查单个用户@app.get("/user/{user_id}")defget_user(user_id:int,db=Depends(get_db)):# select + whereuser=db.scalar(select(User).where(User.id==user_id))user=db.get(User,user_id)# 主键查询最快,最简单ifnotuser:return{"msg":"用户不存在"}return{"user":user}# 5. 添加用户@app.post("/user/add")defadd_user(username:str,password:str,db=Depends(get_db)):new_user=User(username=username,password=password)db.add(new_user)return{"msg":"添加成功"}# 6. 修改用户@app.post("/user/update")defupdate_user(user_id:int,password:str,db=Depends(get_db)):user=db.scalar(select(User).where(User.id==user_id))ifnotuser:return{"msg":"用户不存在"}# 直接修改user.password=passwordreturn{"msg":"修改成功"}# 7. 删除用户@app.delete("/user/delete/{user_id}")defdelete_user(user_id:int,db=Depends(get_db)):user=db.scalar(select(User).where(User.id==user_id))ifnotuser:return{"msg":"用户不存在"}db.delete(user)return{"msg":"删除成功"}

问题2

sqlalchemy 和 pymysql 是干嘛的?
- sqlalchemy: 封装了数据库操作,不用手写复杂SQL
- pymysql: 是连接python和MySQL的驱动(默认同步操作数据库)
- aiomysql: 是异步数据库驱动(异步操作数据库)
server_default 和 insert_default 的区别?
- server_default -> 数据库默认值
- insert_default -> python插入数据时默认值
create_engine(DB_URL) 和 db = SessionLocal() 的区别?
- create_engine(DB_URL) -> 创建连接引擎-锁定数据库
- db = SessionLocal() -> 创建会话连接-真正连接到数据库(增删改查)
yield 和 Depends() 的作用?
- yield: 借出一个真正的连接器
- Depends(): 获取依赖项,注入路由处理函数参数
db.execute 和 db.scalar / db.scalars 的区别?
- db.execute(sql) -> 底层全能方法能执行所有SQL语句(复杂), 但返回的是原始结果集(ORM对象)
- db.scalar / db.scalars -> 查询的快捷方法, 自动提取数据
db.快捷方法有哪些?
- 查单个（主键）：db.get(模型, id)
- 查单个（条件）：db.scalar(select(模型).where(条件))
- 查多个：db.scalars(select(模型)).all()
- 增单个：db.add(对象)
- 增批量：db.add_all(列表)
- 删单个：db.delete(对象)
- 提交：db.commit()
- 回滚：db.rollback()
查询的用法

# 新增导入：排序、函数、多条件(与或非)fromsqlalchemyimportselect,func,desc,asc,and_,or_,not_ result=db.scalars(select(模型).where(条件).order_by(排序).offset(跳过).limit(条数)).all()