嘘~ 正在从服务器偷取页面 . . .

FastApi


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)
    • 注意 重新加载 会导致模块重复导入 可能是创建了一个新的进程进行监听 导致的问题

  1. 收到无效请求 报错 Invalid HTTP request received
    • pip install wsproto 然后启动命令增加 --ws wsproto
    • uvicorn app:application --host 0.0.0.0 --no-access-log --ws wsproto --port 5555
  2. 报错 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 并不是必须的 同步函数也可以执行
  • 你可以返回一个 dictlist,像 strintPydantic 一样的单个值,等等。
  • 还有许多其他将会自动转换为 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 导入的

  • 您可以使用 Configschema_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是「装饰器」方法(getpost 等)的一个参数。不像之前的所有参数和请求体,它不属于路径操作函数
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 是「装饰器」方法(getpost 等)的一个参数。不像之前的所有参数和请求体,它不属于路径操作函数
  • 你可以使用来自 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):把 datastrbytes)写入文件;

    • read(size):按指定数量的字节或字符读取文件内容;

    • seek(offset):移动至文件 offsetint)字节处的位置

      • 例如,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 摘要 在 docsurl 后方 会显示此处的信息

  • 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这个接口
  • 通常,令牌设置为在一段时间后过期。
    • 因此,用户将不得不在稍后的某个时刻再次登录。
    • 如果令牌被盗,风险就会降低。它不像一个永久的钥匙,将永远工作(在大多数情况下)。
  • 登录成功后 就可以进行操作

创建用户认证

  1. 定义多重依赖 其中一个函数是用来对传过来的账号密码进行验证
from fastapi.security import OAuth2PasswordBearer

# 创建一个 auto2 来验证账号 后面的 /api/LoginVerify 代表表单请求提交的地址
oauth2_scheme = OAuth2PasswordBearer(tokenUrl='/api/LoginVerify')
# 注意 除了下面那个 verify_token 函数以外 不要让任何验证依赖于 这个 oauth2_scheme 否则随便填什么都可以验证通过
  1. 创建一个函数 专门用来验证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
  1. 添加一个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"} # 如果验证失败 响应格式按照规范需要是这样
        )
  1. 上面调用的那个调用的创建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」类。
  • 所有相同的选项都得到支持。
  • 所有相同的 parametersresponsesdependenciestags 等等。
  • 在此示例中,该变量被命名为 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 还以为是出了什么错误
  • 后面发现是日志模块 自动传递给父类 再次打印出来

解决方法

  1. 如果你想关闭 uvicorn 的日志 log_config=None 增加到启动配置文件里面即可
    • 如果你设置了 log_config 其他的日志记录会被禁用
  2. 如果想关闭 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

  • 上面的还没解决可以考虑一下换台电脑

    1. 我电脑是笔记本 win11 用的 wifi 然后我把代码原封不动的放在 公司电脑 台式 win10 接网线就不报错了 稳得很
    2. 可以发现是系统底层的问题
    3. 刚刚尝试了 把笔记本拿过去接 网线 还是报错 说明是底层问题 晚上切换 windwos10
    4. 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="到达信息")
  • 后面的括号没掉了 变成了类对象 而不是 实例 但启动却不报错 访问网页接口文档的时候 才直接报错

文章作者: 林木木
版权声明: 本博客所有文章除特別声明外,均采用 CC BY 4.0 许可协议。转载请注明来源 林木木 !
评论
  目录