Fastapi框架入门
正常情况下 服务器异常了 就没有对应的错误信息,无法从响应里面获取具体的错误信息,可以进行自定义把下面捕获服务端异常的代码放在app实例化之后就行#捕获服务端异常的信息content={},@router.post("/error2",tags=["错误处理"],summary="服务端异常")a = 1/0raise HTTPException(status_code=500,detail="服
后端基础
一、参数类型校验
- 以前类型提示没有强制作用,传递错误的数据,只会有提示,调用的时候不会报错
def add(a:int,b:int):
return a + b
print(add(a:"1",b:"2"))
- 定义接口的时候,这样是不行的,需要对参数进行类型校验,强制验证,不符合直接报错,需要先安装pydantic库
pip install pydantic -i https://mirrors.aliyun.com/pypi/simple
# 安装参数类型校验的模块,可以不安装这个,直接安装fastapi 自带的
- 使用pydantic
- 使用pydantic先定义一个类,里面写上你需要的参数,以及参数类型
- 定义函数的时候,参数的类型直接引用这个类
- 补充:dict方法可以将类实例转化成字典
- 特殊情况:参数少了会报错,参数多了不会报错
- 特殊情况:要求为整数,但是传入了整数字符串,pydantic会自动转换
from pydantic import BaseMode class AddParams(BaseModel): a:int b:int def add(parameter:AddParams): res = parameter.a + parameter.b return res def add2(parameter:AddParams): return parameter if __name__ == '__main__': res = add2(AddParams(a=1, b=2)) print(res.dict()) print(type(res))
二、FastApi基础
1、安装uvicorn 插件
当前阶段uvicorn插件的作用就是驱动FastApi应用代码运行
pip install fastapi uvicorn -i https://mirrors.aliyun.com/pypi/simple
2、导入FasTAPI类
创建一个FastAPI应用程序实例,再定义路由和视图函数,路由是一个字符串,表示我们需要绑定到该路由的URL路径
视图函数是一个Python函数,当该路由收到请求时会被调用,就是后端内部的处理逻辑,比如对数据库的增删改查,进行消息队列mq的生产与消费
一个完整的请求需要包含:请求方式,请求路径,请求参数(get请求不需要),请求头信息
import uvicorn
from fastapi import FastAPI
#实例化FastApi
app = FastAPI()
#定义接口,需要用实例化的FastApi去装饰
@app.get("/hello") #/hello 为接口路径,实际完整地址:【http://127.0.0.1:8000/hello】
def say_hi():
return "这是第一个FasstApi接口" #return 代表接口响应信息
#启动接口服务
if __name__ == '__main__':
uvicorn.run(app, host="127.0.0.1", port=8000)

3、FastApi在线Swagger接口文档
网站为自己定义的接口地址+/docs,这个接口文档不一定能打开,因为需要访问外网
一个解决办法为,下载static.rar压缩包,再在定义的时候加上下面这段代码就可以打开了,需要解压,任何把文件放在接口同级

from fastapi import FastAPI
import uvicorn
from fastapi.openapi.docs import get_swagger_ui_html
from starlette.staticfiles import StaticFiles
app = FastAPI(docs_url=None)
# 挂载静态文件路径
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
return get_swagger_ui_html(
openapi_url=app.openapi_url,
title="接口文档",
swagger_js_url="/static/swagger-ui-bundle.js",
swagger_css_url="/static/swagger-ui.css",
swagger_favicon_url="/static/favicon.png",
swagger_ui_parameters={"defaultModelsExpandDepth": -1},
)
4、补充:async 协程,python中节约资源的一种方式
主要用于异步请求
python中有进程、线程、协程
- 进程:一个程序就是一个进程,有自己独立的空间,资源共享,占用资源多
- 线程:一个进程里面可以有多个线程,资源消耗少一点
- 协程:一个线程里面可以有多个协程
5、查询参数
查询参数是拼接到url地址里面 ?xx=xx&xxx=xxx
在FastAPI里面,无论是get还是post请求,如果没有使用pydantic进行验证,直接写的参数,都是查询参数
import uvicorn
from fastapi import FastAPI
from fastapi.openapi.docs import get_swagger_ui_html
from pydantic import BaseModel, Field
from starlette.staticfiles import StaticFiles
app = FastAPI(docs_url=None)
# 挂载静态文件路径
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
return get_swagger_ui_html(
openapi_url=app.openapi_url,
title="接口文档",
swagger_js_url="/static/swagger-ui-bundle.js",
swagger_css_url="/static/swagger-ui.css",
swagger_favicon_url="/static/favicon.png",
swagger_ui_parameters={"defaultModelsExpandDepth": -1},
)
class LoginParams(BaseModel):
username:str=Field(description="用户名",min_length=6,max_length=20)
password:str=Field(description="密码",min_length=6,max_length=20)
@app.post("/login")
def user_login(logindata:LoginParams):
return {"code":200,"message": f"{logindata.username}登录成功,密码:{logindata.password}"}
if __name__ == '__main__':
uvicorn.run(app)
6、路径参数
不同的商品 前面的部分是一样的 最后的数字是变化的,直接写在斜杠后面 这个就是路径参数

import uvicorn
from fastapi import FastAPI
from fastapi.openapi.docs import get_swagger_ui_html
from pydantic import BaseModel, Field
from starlette.staticfiles import StaticFiles
app = FastAPI(docs_url=None)
# 挂载静态文件路径
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
return get_swagger_ui_html(
openapi_url=app.openapi_url,
title="接口文档",
swagger_js_url="/static/swagger-ui-bundle.js",
swagger_css_url="/static/swagger-ui.css",
swagger_favicon_url="/static/favicon.png",
swagger_ui_parameters={"defaultModelsExpandDepth": -1},
)
@app.get("/user/{name}")
def get_user(name: str,age:int):
return {"code":200,"message":f"{name}你好,今年{age}岁"}
if __name__ == '__main__':
uvicorn.run(app)
7、APIRouter的使用(接口整合)
- 现在每个py文件的接口都是独立的,每次写接口的时候,都要指定本地文件,接口文档显示的接口都要复制粘贴
- 解决办法:使用FastAPI的APIRouter定义接口路由,使用统一的启动入口,入口写一次实例化指向本地文件,接口文件显示的接口
接口需要直接使用app=FastAPI()进行定义,而是使用router=APIRouter()进行定义
from fastapi import APIRouter
#创建一个APIRouter实例化
router= APIRouter()
@router.get("/hello")
def hello():
return "这是使用APIRouter定义的接口"
运行入口
import uvicorn
from fastapi import FastAPI
from fastapi.openapi.docs import get_swagger_ui_html
from starlette.staticfiles import StaticFiles
from apirouter的使用_07 import router as r1
app = FastAPI(docs_url=None)
# 挂载静态文件路径
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
return get_swagger_ui_html(
openapi_url=app.openapi_url,
title="接口文档",
swagger_js_url="/static/swagger-ui-bundle.js",
swagger_css_url="/static/swagger-ui.css",
swagger_favicon_url="/static/favicon.png",
swagger_ui_parameters={"defaultModelsExpandDepth": -1},
)
app.include_router(r1)
if __name__ == '__main__':
uvicorn.run(app)
8、接口的分组和说明
定义接口的时候,在路由后面加上tags和summary就可以进行分组和注释了
from fastapi import APIRouter
from pydantic import BaseModel, Field
router = APIRouter()
class UserData(BaseModel):
age:int=Field(description="年龄",ge=18,le=100)
name:str=Field(description="姓名",min_length=2,max_length=10)
@router.post("/user",tags=["用户管理"],summary="用户添加接口")
def user_add(params:UserData):
return {"code":200,"message":f"{params.name}添加成功,年龄为:{params.age}"}

9、自定义响应状态码
正常情况下 服务器异常了 就没有对应的错误信息,无法从响应里面获取具体的错误信息,可以进行自定义
把下面捕获服务端异常的代码放在app实例化之后就行
#捕获服务端异常的信息
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
stack_trace = traceback.format_exc().split("\n")[-5:]
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={
"message": "Internal Server Error",
"stack_trace": stack_trace,
},
)
from fastapi import APIRouter,HTTPException
router = APIRouter()
@router.post("/error2",tags=["错误处理"],summary="服务端异常")
def error2():
a = 1/0
raise HTTPException(status_code=500,detail="服务器内部错误")

10、依赖注入
Depend是设置参数的依赖,一些数据可能会有其他依赖,这种情况下可以使用Depends来实现
from fastapi import APIRouter,Depends
#定义一个被依赖的接口
def login(acc:str,pwd:str):
if acc == "admin" and pwd == "123456":
return True
else:
return False
router = APIRouter()
@router.post("/order")
def order(login_status:bool = Depends(login),item_id:int=None):
print(f"login_status:{login_status}")
return {"code":200,"message":f"{item_id}下单成功"}

更多推荐


所有评论(0)