FastApi框架
目录
一、定义
二、文档网址
三、简介特点
3.1简介
3.2特点
3.3适用场景
四、简单使用
五、进阶使用方法
5.1路由构建方法
5.2数据请求和响应模型
5.2.1数据请求模型
5.2.2响应模型
5.2.3代码实现
5.2.4界面分析
六、习题
一、定义
FastAPI 是一个用于构建 API 的现代、快速(高性能)的 web 框架,使用 Python 并基于标准的 Python 类型提示。
二、文档网址
fastapi文档网址:FastAPI
三、简介特点
3.1简介
FastAPI 是一个用于构建 API 的现代、快速(高性能)的 web 框架,专为在 Python 中构建 RESTful API 而设计。
FastAPI 使用 Python 3.8+ 并基于标准的 Python 类型提示。
FastAPI 建立在 Starlette 和 Pydantic 之上,利用类型提示进行数据处理,并自动生成API文档。
3.2特点
高性能: 基于Starlette和Pydantic,利用异步(asynchronous)编程,提供出色的性能。
自动文档生成: 自动生成交互式API文档,支持Swagger UI和ReDoc,让API的理解和测试更加直观。
类型注解支持: 利用Python的类型提示,提供更严格的输入验证和更好的代码提示。
异步支持: 支持异步请求处理,使得处理IO密集型任务更加高效。
3.3适用场景
用于构建RESTful API,支持前后端分离的Web应用。可以作为微服务的后端框架,支持快速开发和部署
四、简单使用
第一步:导入fastapi库
第二步:创建fastapi实例
第三步:定义根路径 / 的路由
第四步:导入uvicorn库,创建路由和端口
from fastapi import FastAPI
from typing import Unionapp = FastAPI(title='用户信息系统', description='用户信息管理系统', version='1.0.1')@app.get("/") #根路由地址
async def root():return {"message": "Hello World"}if __name__ == "__main__":import uvicornuvicorn.run(app, host="127.0.0.1", port=8000)
运行代码,控制台会出来一个地址
点击地址进入:
在上面的网址结尾加上/docs进入交互式api文档
一个简单的FastApi就够建好了
五、进阶使用方法
5.1路由构建方法
get:获取元素信息
post:添加元素信息
put:更改元素信息
delete:删除元素信息
from fastapi import FastAPI
from typing import Unionapp = FastAPI(title='用户信息系统', description='用户信息管理系统', version='1.0.1')@app.get("/") #根路由地址
async def root():return {"message": "Hello World"}@app.get('/userinfo/{username}/{id1}', tags=['查询用户信息'])
async def get_message(username: str, id1: int):return {'message': '用户信息','username': username,'id': id1}@app.post('/users', tags=['添加用户信息'])
async def create_user(user_data: dict):# 这里应该实现实际的用户创建逻辑return {"message": "用户创建成功", "user": user_data}@app.put('/users/{user_id}', tags=['更新用户信息'])
async def update_user(user_id: int, user_data: dict):# 这里应该实现实际的用户更新逻辑return {"message": f"用户 {user_id} 更新成功", "user": user_data}@app.delete('/users/{user_id}', tags=['删除用户信息'])
async def delete_user(user_id: int):# 这里应该实现实际的用户删除逻辑return {"message": f"用户 {user_id} 删除成功"}if __name__ == "__main__":import uvicornuvicorn.run(app, host="127.0.0.1", port=8000)
5.2数据请求和响应模型
5.2.1数据请求模型
定义:
请求模型用于定义客户端发送给服务器的数据结构,使用 Pydantic 的 BaseModel 类创建模型
适用场景:
(1)通过请求体(Request Body)接收数据
(2)自动验证和解析客户端发送的 JSON 数据
(3)提供数据类型检查和错误提示
5.2.2响应模型
定义:
响应模型用于定义服务器返回给客户端的数据结构:同样使用 Pydantic 的 BaseModel
使用方式:
(1)通过 response_model 参数指定
(2)自动序列化返回数据为 JSON
(3)过滤掉未在模型中定义的字段
5.2.3代码实现
from fastapi import FastAPI
from typing import Union
from pydantic import BaseModelapp = FastAPI(title='用户信息系统', description='用户信息管理系统', version='1.0.1')class User(BaseModel):username: strpassword: stremail: strphone: str@app.get("/") #根路由地址
async def root():return {"message": "Hello World"}#请求模型
@app.post('/users', tags=['添加用户信息'])
async def create_user(user_data: User):return {"code": 200,"message": "用户创建成功","data": user_data}#响应模型
@app.post('/users/response', tags=['添加用户信息'], response_model=User)
async def create_user_response(user_data: User):return user_dataif __name__ == "__main__":import uvicornuvicorn.run(app, host="127.0.0.1", port=8000)
5.2.4界面分析
请求模型:
请求体默认是以json形式,在里面的数据中填入需要添加的用户信息
响应模型:
六、习题
创建一个联通sqlite数据库,包含用户登陆注册和增删改查的系统
思路:
1.创建sqlite表格users,包含username,password,email,以及id自增的表格
2.fastapi创建路由登陆注册,对用户是否存在进行判断,利用请求模型规定必须要输入的值
3.用户信息的增删改查,从sqlite表中读取数据并且根据指定操作后,返回给表
代码实现如下:
# day0829.py
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.orm import Sessionfrom typing import List, Optionalapp = FastAPI() # 创建一个app应用对象# 数据库配置
DATABASE_URL = "sqlite:///./user.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()# 数据库模型
class User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True, index=True)username = Column(String(50), unique=True, index=True)email = Column(String(100), unique=True, index=True)password = Column(String(100))# 创建表
Base.metadata.create_all(bind=engine)class UserCreate(BaseModel):username: stremail: strpassword: strclass UserLogin(BaseModel):username: strpassword: strclass UserUpdate(BaseModel):email: Optional[str] = Nonepassword: Optional[str] = Noneclass UserResponse(BaseModel):id: intusername: stremail: strclass Config:orm_mode = Truedef get_db():db = SessionLocal()try:yield dbfinally:db.close()@app.get('/')
async def index():return {'message': 'FastAPI ORM操作sqlite数据库 - 用户管理系统'}# 用户注册
@app.post('/register',summary='用户注册', response_model=UserResponse)
async def register_user(user: UserCreate, db: Session = Depends(get_db)):db_user = db.query(User).filter(User.username == user.username).first()if db_user:raise HTTPException(status_code=400, detail="用户名已存在")db_user = db.query(User).filter(User.email == user.email).first()if db_user:raise HTTPException(status_code=400, detail="邮箱已被注册")new_user = User(username=user.username,email=user.email,password=user.password)db.add(new_user)db.commit()db.refresh(new_user)return new_user@app.post('/login',summary='用户登录')
async def login_user(user: UserLogin, db: Session = Depends(get_db)):db_user = db.query(User).filter(User.username == user.username).first()if not db_user:raise HTTPException(status_code=400, detail="用户不存在")if db_user.password != user.password:raise HTTPException(status_code=400, detail="密码错误")return {"message": "登录成功", "user_id": db_user.id}@app.get('/users',summary='查询全部用户信息', response_model=List[UserResponse])
async def get_users(db: Session = Depends(get_db)):users = db.query(User).all()return users@app.get('/users/{user_id}',summary='通过id查询信息', response_model=UserResponse)
async def get_user(user_id: int, db: Session = Depends(get_db)):db_user = db.query(User).filter(User.id == user_id).first()if not db_user:raise HTTPException(status_code=404, detail="用户不存在")return db_user@app.put('/users/{user_id}',summary='更新用户信息', response_model=UserResponse)
async def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):db_user = db.query(User).filter(User.id == user_id).first()if not db_user:raise HTTPException(status_code=404, detail="用户不存在")# 更新用户信息if user.email is not None:existing_user = db.query(User).filter(User.email == user.email, User.id != user_id).first()if existing_user:raise HTTPException(status_code=400, detail="邮箱已被其他用户使用")db_user.email = user.emailif user.password is not None:db_user.password = user.passworddb.commit()db.refresh(db_user)return db_user@app.delete('/users/{user_id}',summary='删除用户信息')
async def delete_user(user_id: int, db: Session = Depends(get_db)):db_user = db.query(User).filter(User.id == user_id).first()if not db_user:raise HTTPException(status_code=404, detail="用户不存在")db.delete(db_user)db.commit()return {"message": "用户删除成功"}if __name__ == '__main__':import uvicornuvicorn.run('day0829zuoye:app', host='127.0.0.1', port=8000, reload=True)
总框架:
经过测试后的user表格: