FastApi
- 安装
pip install fastapi[all]
使用这个指令会安装所需依赖项 - 以上安装还包括了
uvicorn
你可以将其用作运行代码的服务器 uvicorn main:app --reload --port 80
开启本地服务器- 其中
main
指定main.py
这个文件 - 其中
app
是某个py
文件中FastAPI
的实例名 uvicorn --help
获取所有帮助
- 其中
- 自带交互式文档
http://127.0.0.1/docs
- 可选
api
文档http://127.0.0.1/redoc
代码层启动
:uvicorn.run(app="Server:app", host="0.0.0.0", port=Settings.ServerPort, log_level="info", debug=Settings.DEBUG, use_colors=True)
- 注意
重新加载
会导致模块重复导入 可能是创建了一个新的进程进行监听 导致的问题
- 注意
- 收到无效请求 报错
Invalid HTTP request received
pip install wsproto
然后启动命令增加--ws wsproto
uvicorn app:application --host 0.0.0.0 --no-access-log --ws wsproto --port 5555
- 报错
Unsupported upgrade request.
pip uninstall uvicorn
pip install uvicorn[standard]
启动方式
- 因为我需要用
nacos
每次启动就会报错 我就用下面的方式启动 会关掉自动重载 - 我这里设置
reload=False
还是会自动重载
async def main():
config = uvicorn.Config(
app="app:app",
host="0.0.0.0",
port=Settings.ServerPort,
log_level="debug",
use_colors=True,
reload=False,
debug=False,
)
server = uvicorn.Server(config)
await server.serve()
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.run_forever()
简单用法
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
# 控制台执行 即可
# uvicorn RunServer:app --reload --port 80
- 其中
async
并不是必须的 同步函数也可以执行 - 你可以返回一个
dict
、list
,像str
、int
、Pydantic
一样的单个值,等等。 - 还有许多其他将会自动转换为 JSON 的对象和模型(包括 ORM 对象等)。
路径参数
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
- 注意 这里在
item_id
后面指明了类型int
如果前端传过来的是字符串的数字类型 也会被自动转换- 如果前端传的不是数字类型 或者 不能被转换为数字类型 则会得到一个正确的报错
float
会被强制转换为int
导致精度丢失
- 如果你定义了两个相同的路径参数
/items/{item_id}
和/items/me
那么调用顺序就是定义顺序 - 你还可以事先定义数据 为 枚举类型
class ModelName(str, Enum):
alexnet = "alexnet"
resnet = "resnet"
lenet = "lenet"
@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
pass
- 通过从
str
继承,API 文档将能够知道这些值必须为string
类型并且能够正确地展示出来。 - 你可以使用
model_name.value
或通常来说your_enum_member.value
来获取实际的值(在这个例子中为str
)
文件地址路径参数
- 因为路径中 有
/
会导致路径错误 那么可以使用 Starlette 的一个内部工具在 FastAPI 中实现它。
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
return {"file_path": file_path}
查询字符串
- 查询字符串是键值对的集合,这些键值对位于 URL 的
?
之后,并以&
符号分隔。
list_data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
@app.get("/")
async def root(start: int = 0, end: int = 10) -> list_data:
return list_data[start:end]
# http://127.0.0.1/?start=6&end=9
- 可以访问到指定列表的内容 有默认值 并且进行数据校验
- 如果你不想添加一个特定的值,而只是想使该参数成为可选的,则将默认值设置为
None
。 - 但当你想让一个查询参数成为必需的,不声明任何默认值就可以
查询参数 与 字符串校验
- 现在已经可以指定参数的类型 和 默认值 当然也可以指定数据的 校验规则
@app.get("/items/")
async def read_items(
q: Union[str, None] = Query(default='start_hello_end', min_length=3, max_length=50, regex="^start.*?end$")
):
return q
- 你还可以指定参数为多次出现的值 如
http://localhost:8000/items/?q=foo&q=bar
- 那么定义的时候就要使用
q: Union[List[str], None]
来进行接收
别名参数
假设你想要查询参数为
item-query
。像下面这样:http://127.0.0.1:8000/items/?item-query=foobaritems
但是
item-query
不是一个有效的 Python 变量名称。最接近的有效名称是
item_query
。但是你仍然要求它在 URL 中必须是
item-query
这时你可以用
alias
参数声明一个别名,该别名将用于在 URL 中查找查询参数值:
@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, alias="item-query")):
results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
if q:
results.update({"q": q})
return results
- 如果未指定别名 就不会接收到 参数
q
弃用参数
- 现在假设你不再喜欢此参数。
- 你不得不将其保留一段时间,因为有些客户端正在使用它,但你希望文档清楚地将其展示为已弃用。
- 那么将参数
deprecated=True
传入Query
即可
路径参数 与 字符串检验
- 与使用
Query
为查询参数声明更多的校验和元数据的方式相同,你也可以使用Path
为路径参数声明相同类型的校验和元数据。 - 上面使用
Query
对查询参数进行校验 这里使用path
对路径参数进行校验
# 数值校验:大于和小于等于¶
# 同样的规则适用于:
#
# gt:大于(greater than)
# le:小于等于(less than or equal)
@app.get("/items/{item_id}")
async def read_items(
item_id: int = Path(title="The ID of the item to get", gt=0, le=1000),
):
return item_id
请求体
- 请求体一样可以定义为模型 也推荐这样做
class Item(BaseModel):
name: str
description: Union[str, None] = None # 可选 str 或者 None
price: float
tax: Union[float, None] = None # 可选 str 或者 None
@app.post("/items/")
async def create_item(item: Item):
return item
仅仅使用了 Python 类型声明,FastAPI 将会:
- 将请求体作为 JSON 读取。
- 转换为相应的类型(在需要时)。
- 校验数据。
- 如果数据无效,将返回一条清晰易读的错误信息,指出不正确数据的确切位置和内容。
- 将接收的数据赋值到参数
item
中。- 由于你已经在函数中将它声明为
Item
类型,你还将获得对于所有属性及其类型的一切编辑器支持(代码补全等)。
- 由于你已经在函数中将它声明为
- 为你的模型生成 JSON 模式 定义,你还可以在其他任何对你的项目有意义的地方使用它们。
- 这些模式将成为生成的 OpenAPI 模式的一部分,并且被自动化文档 UI 所使用。
- 你还可以同时声明请求体、路径参数和查询参数。
- FastAPI 会识别它们中的每一个,并从正确的位置获取数据。
获取请求体 与 设置响应信息
- 这里的请求不是用户发来的信息
- 而是浏览器自动带上的 比如 你的另外一个请求设置的
cookie
- 或者你希望给前端设置一个
Cookie
from fastapi import APIRouter, Response, Request
@Router.post(path="/LoginVerify/", response_model=ReturnModel)
async def login_verify(request: Request, responses: Response, request_data: LoginVerifyModel):
"""登录验证接口"""
print(request.cookies.get('token'))
if request_data.password in Login_Password:
responses.set_cookie("token", request_data.password)
return ReturnModel(success=True, code=HTTP_200_OK, message='登录成功')
else:
return ReturnModel(success=False, code=HTTP_200_OK, message='登录失败')
- 如果觉得导入整个
response
麻烦 可以直接导入Cookie
@Router.post(path="/LoginVerify/", response_model=ReturnModel)
async def login_verify(request_data: LoginVerifyModel, response: Response, token: str = Cookie(default="")):
"""登录验证接口"""
print(token)
- 根据参数名 会自动获取前端的参数值
@app.get("/items/")
async def read_items(user_agent: Union[str, None] = Header(default=None)):
return {"User-Agent": user_agent}
通过 pydantic 校验
- 这里只简单的使用提一下
pydantic
后面单独开一篇 文章 学习pydantic
字段
class Item(BaseModel):
name: str
description: Union[str, None] = Field(
default=None, title="The description of the item", max_length=300
)
price: float = Field(gt=0, description="The price must be greater than zero")
tax: Union[float, None] = None
- 注意,
Field
是直接从pydantic
导入的
- 您可以使用
Config
和schema_extra
为 Pydantic 模型声明一个示例.
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
class Config:
schema_extra = {
"example": {
"name": "Foo",
"description": "A very nice Item",
"price": 35.4,
"tax": 3.2,
}
}
响应模型
- 注意,
response_model
是「装饰器」方法(get
,post
等)的一个参数。不像之前的所有参数和请求体,它不属于路径操作函数。
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
tags: List[str] = []
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
return item
FastAPI 将使用此 response_model
来:
- 将输出数据转换为其声明的类型。
- 校验数据。
- 在 OpenAPI 的路径操作中为响应添加一个 JSON Schema。
- 并在自动生成文档系统中使用。
但最重要的是:
- 会将输出数据限制在该模型定义内。下面我们会看到这一点有多重要。
响应状态码
@app.post("/items/", status_code=201)
async def create_item(name: str):
return {"name": name}
- 注意,
status_code
是「装饰器」方法(get
,post
等)的一个参数。不像之前的所有参数和请求体,它不属于路径操作函数。 - 你可以使用来自
fastapi.status
的便捷变量。
from fastapi import FastAPI, status
app = FastAPI()
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str):
return {"name": name}
- 这里的响应表示成功响应的状态码 并不是 自定义响应状态码
表单字段
- 接收的不是 JSON,而是表单字段时,要使用
Form
。
@app.post("/login/")
async def login(username: str = Form(), password: str = Form()):
return {"username": username}
Form
是直接继承自Body
的类。
@app.post("/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
# 这里获取到 用户名密码 验证之后才可以登录成功
return {"access_token": "admin", "token_type": "bearer"}
File 上传文件
- 首先需要安装
pip install python-multipart
- 如果是用
[all]
进行的安装FastApi
那么就不需要进行这一步操作
- 如果是用
@app.post("/files/")
async def create_file(file: bytes = File()):
return {"file_size": len(file)}
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile):
return {"filename": file.filename}
上面展示了两种方式 很多情况下,
UploadFile
更好用。UploadFile
的属性如下:filename
:上传文件名字符串(str
),例如,myimage.jpg
;content_type
:内容类型(MIME 类型 / 媒体类型)字符串(str
),例如,image/jpeg
;file
: 其实就是 Python 文件,可直接传递给其他预期file-like
对象的函数或支持库。UploadFile
支持以下async
方法,(使用内部SpooledTemporaryFile
)可调用相应的文件方法。write(data)
:把data
(str
或bytes
)写入文件;read(size)
:按指定数量的字节或字符读取文件内容;seek(offset):移动至文件
offset
(int
)字节处的位置- 例如,
await myfile.seek(0)
移动到文件开头; - 执行
await myfile.read()
后,需再次读取已读取内容时,这种方法特别好用;
- 例如,
close()
:关闭文件。
因为上述方法都是
async
方法,要搭配「await」使用。例如,在
async
路径操作函数 内,要用以下方式读取文件内容:contents = await myfile.read()
在普通
def
路径操作函数 内,则可以直接访问UploadFile.file
,例如:contents = myfile.file.read()
当然也支持多文件上传
处理异常
- 向客户端返回 HTTP 错误响应,可以使用
HTTPException
。
@Router.put("/AddPerson/")
async def person(person_data: VerifyPerson):
PersonSession.add(SqlPerson(**person_data.dict()))
try:
PersonSession.commit()
except IntegrityError as e:
PersonSession.rollback()
raise HTTPException(status_code=HTTP_200_OK, detail=f"add error -> {e.args}")
finally:
PersonSession.close()
return ReturnBaseModel(success=True, code=HTTP_200_OK, message='添加成功')
- 并且可以添加许多自定义信息 头 等.
- 在控制台输出的 异常类型不准确 需要用
type
函数输出准确的类型 才可以准确的拦截下来 - 可以用
finally
来保证关闭连接
装饰器额外参数
- 注意:以下参数应直接传递给路径操作装饰器,不能传递给路径操作函数。
- 也就是 这里的参数 只能传递给
@app.get("/items/{item_id}") # 这个部分的装饰器
async def read_item(item_id: str):
status_code
用于定义路径操作响应中的 HTTP 状态码。- 可以直接传递
int
代码, 比如404
。 - 如果记不住数字码的涵义,也可以用
status
的快捷常量:
- 可以直接传递
tags
参数的值是由str
组成的list
(一般只有一个str
)- 用于为路径操作添加标签
- 在
docs
中会根据这个标签 进行接口的分类
summary
摘要 在docs
的url
后方 会显示此处的信息description
简介 在摘要的下方增加一个简介docstring
文档字符串 这个字符串在函数内第一个的多行字符串 列如
async def read_item(item_id: str):
"""
这是文档字符串简介
"""
return {"item": items[item_id]}
# 如果同事存在 文档字符串 和 description 简介 那么就会以 description 为准进行显示
response_description
响应描述deprecated
将路径标注为弃用
兼容的 JSON
- 在某些情况下,您可能需要将数据类型(如 Pydantic 模型)转换为与 JSON 兼容的内容(如 、 等)。
dict``list
- 例如,如果需要将其存储在数据库中。
- 为此,FastAPI提供了一个功能。
jsonable_encoder()
from fastapi.encoders import jsonable_encoder
class Item(BaseModel):
title: str
timestamp: datetime
description: Union[str, None] = None
app = FastAPI()
@app.put("/items/{id}")
def update_item(id: str, item: Item):
json_compatible_item_data = jsonable_encoder(item)
fake_db[id] = json_compatible_item_data
依赖系统
函数依赖项
async def two(s: str) -> str:
return s * 2
@app.get("/items/{item_id}")
async def read_item(item_id: str = Depends(two)):
return item_id
- 通过
Depends
表明item_id
依赖于tow
其中two
函数除了没有@app.get
这类装饰器以外 其他参数都可以用 - 当函数被调用后 传入的参数
item_id
会被传递给tow
函数 并且接受返回值 参数会赋值给item_id
多重依赖项
def query_extractor(q: Union[str, None] = None):
return q
def query_or_cookie_extractor(
q: str = Depends(query_extractor),
last_query: Union[str, None] = Cookie(default=None),
):
if not q:
return last_query
return q
@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
return {"q_or_cookie": query_or_default}
- 这样就申明为多重依赖项
路径操作依赖项
- 有时,我们并不需要在路径操作函数中使用依赖项的返回值。或者说,有些依赖项不返回值。
- 但仍要执行或解析该依赖项。对于这种情况,不必在声明路径操作函数的参数时使用
Depends
,而是可以在路径操作装饰器中添加一个由dependencies
组成的list
。
async def verify_token(x_token: str = Header()):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=HTTP_200_OK, detail="X-Token header invalid")
async def verify_key(x_key: str = Header()):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=HTTP_200_OK, detail="X-Key header invalid")
return x_key
# dependencies=[Depends(verify_token), Depends(verify_key)] 指定依赖项 但是没有获取返回值
@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
return [{"item": "Foo"}, {"item": "Bar"}]
全局依赖项
- 有时,我们要为整个应用添加依赖项。
- 通过与定义路径装饰器依赖项类似的方式,可以把依赖项添加至整个
FastAPI
应用。 - 这样一来,就可以为所有路径操作应用该依赖项:
async def verify_token(x_token: str = Header()):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="X-Token header invalid")
async def verify_key(x_key: str = Header()):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="X-Key header invalid")
return x_key
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
多重依赖 启用用户认真
- 对每个接口进行限制 允许的用户才可以进行登录
# 下面生成一个 token 返回给前端 格式要求固定的
@Router.post(path="/***", response_model=dict)
async def login_verify(request_data: OAuth2PasswordRequestForm = Depends()):
"""登录接口"""
# 当用户密码存在于字典中才生成token返回
if request_data.username and request_data.password:
access_token_expires = timedelta(minutes=Settings.ACCESS_TOKEN_EXPIRE_MINUTES or None)
access_token = create_access_token(
data={
'username': request_data.username,
'password': request_data.password,
},
expires_delta=access_token_expires,
)
return {"access_token": access_token, "token_type": "Bearer"}
else:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户未授权",
headers={"WWW-Authenticate": "Bearer"}
)
# token的创建用 jwt 必须传入一个字典
jwt.encode(claims=to_encode, key=SECRET_TOKEN, algorithm=ALGORITHMS.HS256)
# 字典内部如果有 {"exp": expire} exp 参数 有效时间
# 那么在解密的时候 会自动判断是否超过有效时间 则自动报错
# 然后在需要验证的地方 开启多重依赖
Router = APIRouter(
tags=['PersonInterface'],
dependencies=[Depends(verify_user)]
)
def verify_user(token: str = Depends(oauth2_scheme)):
# 在此进行你的验证 最终返回 tok
pass
中间件
- 在每个 请求 和 响应 之前 调用的函数
请求中间件 响应中间件
request
请求对象- all_next 它将接收 request 作为参数.
- 这个函数将 request 传递给相应的 路径操作.
- 然后它将返回由相应的路径操作生成的 response.
- 然后你可以在返回 response 前进一步修改它.
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
response = await call_next(request)
return response
- 一些验证 比如 登录验证 在中间件处理就比较好
后台任务
- 有的时候我们只需要接收前端发来的消息 并且立即返回 但是我们需要后台进行一系列处理
- 这些时间可能很久 不应该让用户进行等待 那么就可以将任务添加到后台进行处理
def write_notification(email: str, message=""):
# 模拟延迟后发送邮件
time.sleep(5)
print("已发送邮件", email, message)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
# 以下代码 将任务添加到后台 并继续执行代码
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "Notification sent in the background"}
安全性
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='login') # 页面上的登录界面 登录后 就会请求到这个URL
@app.post("/login") # login 也就是这里
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
print(form_data.username)
print(form_data.password)
# 这里获取到 用户名密码 验证之后才可以登录成功
return {"access_token": "admin", "token_type": "bearer"}
@app.get('/')
async def test(token: str = Depends(oauth2_scheme)):
return {'hello': token}
token
这里的 tokenUrl 就是当用户在页面的登录界面点击登录之后- 会将用户名密码 通过 Form 表单传输给指定的
url
127.0.0.1:8000/xxx
- 我们随便输入用户名密码, 点击
Authorize
按钮, 发现它报错说Auth Error Error: Not Found
- 这是因为我们还没有实现这个表单格式的登录接口, 查看后台日志可以看到确实也请求了
/xxx
这个接口
- 会将用户名密码 通过 Form 表单传输给指定的
- 通常,令牌设置为在一段时间后过期。
- 因此,用户将不得不在稍后的某个时刻再次登录。
- 如果令牌被盗,风险就会降低。它不像一个永久的钥匙,将永远工作(在大多数情况下)。
- 登录成功后 就可以进行操作
创建用户认证
- 定义多重依赖 其中一个函数是用来对传过来的账号密码进行验证
from fastapi.security import OAuth2PasswordBearer
# 创建一个 auto2 来验证账号 后面的 /api/LoginVerify 代表表单请求提交的地址
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/api/LoginVerify')
# 注意 除了下面那个 verify_token 函数以外 不要让任何验证依赖于 这个 oauth2_scheme 否则随便填什么都可以验证通过
- 创建一个函数 专门用来验证
token
这个token
依赖于上面创建的oauth2_scheme
# 之后所有需要验证的接口 都增加这个依赖 也可以增加全局依赖
async def verify_token(token: str = Depends(oauth2_scheme)) -> str:
"""
token验证 校验登录状态是否过期
返回 token 表示登录状态有效 否则报异常
"""
# 因为客户端不知道token 修改过的 token w
try:
jwt.decode(token=token, key=SECRET_TOKEN, algorithms=ALGORITHMS.HS512)
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户认证异常",
headers={"WWW-Authenticate": "Bearer"}
)
return token
- 添加一个
api
接口 这个地址就是上面那个url
Router = APIRouter(
tags=['*'],
prefix='/api',
include_in_schema=True
)
# 注意 这个接口不要应用任何依赖 用来创建用户的token
@Router.post(path="/LoginVerify", response_model=dict)
async def login_verify(response: Response, request_data: OAuth2PasswordRequestForm = Depends()):
"""登录接口"""
# 当用户密码存在于字典中才生成token返回
if request_data.username in Settings.ADMINS and Settings.ADMINS.get(request_data.username, '') == request_data.password:
# 这里指定token过期时间
access_token_expires = timedelta(minutes=Settings.ACCESS_TOKEN_EXPIRE_MINUTES or None)
# 这里调用函数 create_access_token 创建一个 用户token
access_token = await create_access_token(
data={
'username': request_data.username,
'password': request_data.password,
},
expires_delta=access_token_expires,
)
# 如果验证通过 格式必须是下面这样
response.headers["Bearer"] = access_token
return {"access_token": access_token, "token_type": "Bearer"}
else:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户未授权",
headers={"WWW-Authenticate": "Bearer"} # 如果验证失败 响应格式按照规范需要是这样
)
- 上面调用的那个调用的创建
token
函数 内部会自动检测到期时间
# 创建 jwt token 令牌 需要自己创建第一无二的令牌
# https://jwt.io/ 创建第一无二的
SECRET_TOKEN = "******"
async def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
"""
创建token
"""
to_encode = data.copy()
expire = datetime.utcnow() + expires_delta if expires_delta else datetime.utcnow() + timedelta(minutes=Settings.ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire}) # 过期时间必须是 exp
return jwt.encode(claims=to_encode, key=SECRET_TOKEN, algorithm=ALGORITHMS.HS512)
路由系统
from fastapi import APIRouter
router = APIRouter()
@router.get("/spider/", tags=["spiders"])
async def spider():
return "hello spider"
@router.get("/users/{username}", tags=["spiders"])
async def read_user(username: str):
return {"username": username}
- 你可以将
APIRouter
视为一个「迷你FastAPI
」类。 - 所有相同的选项都得到支持。
- 所有相同的
parameters
、responses
、dependencies
、tags
等等。 - 在此示例中,该变量被命名为
router
,但你可以根据你的想法自由命名。
- 在上面的
tags
中可以发现每个 接口 都有 一个tags
参数 那么就可以提取出来放在APIRouter
中- 并且 所有参数都可以 进行通用配置
- 由于每个路径操作的路径都必须以
/
开头,但不能以/
作为结尾
from fastapi import FastAPI
from Apps.Spiders import SpiderRouter
app = FastAPI()
app.include_router(SpiderRouter)
- 然后在 主函数文件 导入 路由 进行安装
启动与结束函数
- 有的时候希望在
fastapi
服务启动之前执行某个函数 - 在服务结束之后执行某个函数 那么就可以添加以下代码
@app.on_event("startup")
async def startup_event():
# 当开始app之前 会执行这个函数
# 如果要多个 相同操作即可
pass
@app.on_event("shutdown")
def shutdown_event():
# 当app结束之后 h
# 如果要多个 相同操作即可
pass
日志记录
- 在
app
内设置启动执行日志配置 - 关闭冒泡 控制台不会再输出 错误信息都会跑到日志文件中
- 最新的日志都存在于
api.log
中 超过指定大小会自动备份
# 初始化日志模块
@app.on_event('startup')
async def init_logging():
0 if path.exists("logs") else makedirs('logs')
access_logger = logging.getLogger("uvicorn.access")
access_error = logging.getLogger("uvicorn.error")
# 关闭冒泡
access_logger.propagate = False
access_error.propagate = False
# 设置日志格式
handler = logging.handlers.RotatingFileHandler("logs/api.log", mode="a", encoding='utf-8', maxBytes=1024 * 100, backupCount=6)
handler.setFormatter(logging.Formatter("%(asctime)s - %(filename)s - %(lineno)d - %(module)s - %(levelname)s - %(message)s"))
access_logger.addHandler(handler)
access_error.addHandler(handler)
记录
日志重复
- 今天开发的时候遇到一个异常
Started server process [18600]
INFO:uvicorn.error:Started server process [18600]
INFO: Waiting for application startup.
INFO:uvicorn.error:Waiting for application startup.
INFO: Application startup complete.
INFO:uvicorn.error:Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:777 (Press CTRL+C to quit)
INFO:uvicorn.error:Uvicorn running on http://0.0.0.0:777 (Press CTRL+C to quit)
- 启动的时候不管输出什么 都会打印两次
- 当时看到
INFO:uvicorn.error
还以为是出了什么错误 - 后面发现是日志模块 自动传递给父类 再次打印出来
解决方法
- 如果你想关闭
uvicorn
的日志log_config=None
增加到启动配置文件里面即可- 如果你设置了
log_config
其他的日志记录会被禁用
- 如果你设置了
- 如果想关闭
fastAPI
的默认传递情况 增加以下代码即可
logger = logging.getLogger("uvicorn.error")
logger.propagate = False
uvicorn.run(app="Server:app", host="0.0.0.0",, use_colors=True)
服务未找到
- 因为项目原先是用
Django
写的, 现在我想用FastApi
重写- 需要
Nacos
进行注册 但后面发现不是这个问题
- 需要
- 到最后一步之后 内网端口映射给我
777
端口 所以我本地监听的就是777
然后过随机长的时间 就会报错指定的网络名不再可用。
RROR:asyncio:Task exception was never retrieved
future: <Task finished name='Task-25' coro=<IocpProactor.accept.<locals>.accept_coro() done, defined at C:\Users\51482\AppData\Local\Programs\Python\Python38\lib\asyncio\windows_events.py:563> exception=OSError(22, '指定的网络名不再可用。', None, 64, None)>
Traceback (most recent call last):
File "C:\Users\51482\AppData\Local\Programs\Python\Python38\lib\asyncio\windows_events.py", line 566, in accept_coro
await future
File "C:\Users\51482\AppData\Local\Programs\Python\Python38\lib\asyncio\proactor_events.py", line 818, in loop
conn, addr = f.result()
File "C:\Users\51482\AppData\Local\Programs\Python\Python38\lib\asyncio\windows_events.py", line 812, in _poll
value = callback(transferred, key, ov)
File "C:\Users\51482\AppData\Local\Programs\Python\Python38\lib\asyncio\windows_events.py", line 555, in finish_accept
ov.getresult()
OSError: [WinError 64] 指定的网络名不再可用。
- 然后绑定别的
IP
就不出问题了 但是Django
绑定这个777
端口就不会出错 - 目前都不知道是什么原因
- 2022 年 8 月 5 日 11 点 03 分 昨天结果一顿努力 ( 虽然我也不知道做了什么 )
- 出现了一个新的问题
Traceback (most recent call last):
File "C:\Users\f\scoop\apps\python\current\lib\asyncio\proactor_events.py", line 817, in loop
f = self._proactor.accept(sock)
File "C:\Users\f\scoop\apps\python\current\lib\asyncio\windows_events.py", line 545, in accept
self._register_with_iocp(listener)
File "C:\Users\f\scoop\apps\python\current\lib\asyncio\windows_events.py", line 714, in _register_with_iocp
_overlapped.CreateIoCompletionPort(obj.fileno(), self._iocp, 0, 0)
OSError: [WinError 87] 指定参数不正确
完全看不出来是什么错误 但是这次的错误是可以找到的
uvicorn 在 Windows 和 Python 3.8 上重新加载失败 ·第 529 期 ·encode/uvicorn (github.com)
可以发现是异步循环
IOCP
导致的一些问题需要手动指定循环
uvicorn.run("app:app", loop="asyncio")
还有一个选项就是uvloop
上面的还没解决可以考虑一下换台电脑
- 我电脑是笔记本
win11
用的wifi
然后我把代码原封不动的放在 公司电脑台式
win10
接网线
就不报错了 稳得很 - 可以发现是系统底层的问题
- 刚刚尝试了 把笔记本拿过去接 网线
还是报错
说明是底层问题 晚上切换windwos10
- 2022 年 8 月 8 日 09 点 21 分 重装
win10
解决 该死的win11
- 我电脑是笔记本
2022 年 8 月 26 日 14 点 22 分
今天出现了一个小插曲
File "pydantic\schema.py", line 590, in pydantic.schema.model_process_schema
File "pydantic\schema.py", line 631, in pydantic.schema.model_type_schema
File "pydantic\schema.py", line 251, in pydantic.schema.field_schema
File "pydantic\schema.py", line 219, in pydantic.schema.get_field_info_schema
File "pydantic\schema.py", line 996, in pydantic.schema.encode_default
File "pydantic\json.py", line 97, in pydantic.json.pydantic_encoder
TypeError: Object of type 'ModelMetaclass' is not JSON serializable
- 这个信息呢也不知道什么地方错误 百度也没什么答案
- 最后发现是模型默认值出错
- 原本应该是
arrItem: ResItemInfoModel = Field(default=ResItemInfoModel(), description="到达信息")
- 被我写成了
arrItem: ResItemInfoModel = Field(default=ResItemInfoModel, description="到达信息")
- 后面的括号没掉了 变成了类对象 而不是 实例 但启动却不报错 访问网页接口文档的时候 才直接报错