项目的文件结构

如果要构建应用程序或 Web API ,则很少将所有内容都放在一个文件中。

FastAPI提供了一种方便的工具,可在保持所有灵活性的同时构建应用程序。

如果您使用过 Flask ,那将相当于 Flask 的蓝图。

图片

  • app目录包含所有内容。
  • app目录还有一个app/main.py文件
  • 有一个子目录app/routers/

其中两个 __inint__.py 为空文件,仅代表此为一个 python 模块

ApiRouter

假设处理用户相关路由的文件的是子模块 /app/routers/users.py

但仍然希望他是程序的一部分,而不是单独的一个 程序(???include_router()是 fastapi 提供的一种向路由中添加子路由的方式。其不同于Mount直接挂载 App ,而是将子 router 中的路由都拆解出来添加到根 router

导入 ApiRouter 并创建路由

from fastapi import APIRouter

router = APIRouter()

# 这里的 tags 应该是为了给 自动生成的交互式文档分类
@router.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "Foo"}, {"username": "Bar"}]


@router.get("/users/me", tags=["users"])
async def read_user_me():
    return {"username": "fakecurrentuser"}


@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
    return {"username": username}

flask 的蓝图在注册时就已经声明好主路由。我跟喜欢此种方式,至少确保了路由的字符串不会打错( flask 路由: /, me, /<>)

您可以将其APIRouter视为“迷你FastAPI”类。支持所有相同的选项。所有相同的参数,响应,依赖性,标签等。

另一个路由文件 app/routers/items.py

不需要在每个路径操作中都显式地键入/items/和键入:tags=["items"]

from fastapi import APIRouter, HTTPException

router = APIRouter()


@router.get("/")
async def read_items():
    return [{"name": "Item Foo"}, {"name": "item Bar"}]


@router.get("/{item_id}")
async def read_item(item_id: str):
    return {"name": "Fake Specific Item", "item_id": item_id}

@router.put(
    "/{item_id}",
    tags=["custom"],
    responses={403: {"description": "Operation forbidden"}},
        )
async def update_item(item_id: str):
    if item_id != "foo":
        raise HTTPException(status_code=403, detail="You can only update the item: foo")
    return {"item_id": item_id, "name": "The Fighters"}

此处的 主路由不用添加暂且不表(访问应该是:/item/, /item/item_id. 但是 tags 不添加会自动生成码?。 并不会 还是需要在外部导入此 路由时添加。各有利弊吧。个人喜欢 flask 蓝图

入口文件 main

在这里导入和使用类FastAPI

这将是应用程序中将所有内容捆绑在一起的主文件。

导入子路由

from fastapi import Depends, FastAPI, Header, HTTPException
from .routers import items, users

app = FastAPI()

# 拒绝每个 没有 x-token的请求
async def get_token_header(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")

app.include_router(users.router)

app.include_router(
    items.router,
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(get_token_header)],
    responses={404: {"description": "Not found"}},
)

items 的导入

  • 添加了一个前缀的所有路径操作使用参数prefix的值, .前缀不得包含路由最后的 /
  • 添加一个列表,tags该列表将应用于此路由中的所有的路由
  • 添加预定义responses,该预定义也将包含在所有路由操作中
    • 我们可以传递给路径操作装饰器一个参数responses他接收一个字典数据,键值是不同的 HTTP 状态码,内容是不同情况下的返回内容(字典格式)
    • 如果返回内容包含键值model,那么它的作用与response_model相同,指向的内容是 Pydantic 模型。来源
  • 添加一个列表,dependencies( 依赖,钩子)该列表将被添加到路由器中的所有*路由中,并将针对对它们的每个请求执行(过滤?。

避免名称冲突

我们将items直接导入子模块**并使用 items.router,而不是仅导入其变量router

这是因为我们router在子模块中还有另一个变量users

如果我们一个接一个地导入,例如:

from .routers.items import router
from .routers.users import router

则会发生覆盖

多次使用同一路由文件,但不同 prefix ( path 路径

您也可以.include_router()同一路由文件使用不同的前缀多次使用。

例如,在不同前缀下(例如/api/v1/api/latest)使用同一 API 可能会很有用。

后台任务

您可以定义返回响应要运行的后台任务。

这对于在请求后需要进行的操作很有用,但是客户端实际上并不需要在接收响应之前等待操作完成。

如:

  • 执行操作后发送的电子邮件通知:

    • 由于连接到电子邮件服务器并发送电子邮件的过程通常很慢(几秒钟),因此您可以立即返回响应并在后台发送电子邮件通知。
  • 处理数据:

    • 例如,假设您收到的文件必须经过缓慢的处理,您可以返回“已接受”( HTTP 202 )响应,并在后台对其进行处理。

使用 BackgroundTasks

首先,BackgroundTasks路由的操作函数中导入并定义一个参数

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@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"}

FastAPI将为您创建该类的实例对象 BackgroundTasks。并将其作为该函数的参数传递至函数。

添加进后台任务:

background_tasks.add_task(FunctionName, [args, ...])

Takes 和依赖( Dependency ) 一起使用

from typing import Optional
from fastapi import BackgroundTasks, Depends, FastAPI

app = FastAPI()


def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)


def get_query(background_tasks: BackgroundTasks, q: Optional[str] = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q


@app.post("/send-notification/{email}")
async def send_notification(
      email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)
  ):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

元数据和自动文档

您可以在FastAPI应用程序中自定义几种元数据配置:主要为自动生成的 api 文档的各种描述的配置,如 版本,标题,自动文档的位置(网址,禁用文档,....

不做多余追述,参考文档

静态文件

如果需要从 FastApi 提供静态文件的访问:

首先需要安装

pip install aiofiles
from fastapi import FastAPI, staticfiles
# or
# from fastapi.staticfiles import StaticFiles

app = FastAPI()
app.mount('/static', staticfiles.StaticFiles(directory='template'), name="static")

这样就可以通过 /static/FileNmae 访问服务器目录template下静态文件了. name="static"给它FastAPI内部使用的名称?

对象:StaticFiles(directory=None, packages=None, check_dir=True)

  • directory-表示目录路径的字符串或os.Pathlike
  • packages -python 软件包的字符串列表
  • html-以 HTML 模式运行。index.html如果存在这样的文件,则自动为目录加载。
  • check_dir-确保目录在实例化时存在。默认为True

对于不匹配的请求,静态文件将以“ 404 未找到”或“ 405 方法不允许”响应。在 HTML 模式下,如果404.html文件存在,它将显示为 404 响应

接口测试

fastapi 官方文档也提供 有关于测试的介绍,有需要自己去看


相关推荐:

来自系列:fastapi

分类 python下文章:

1.0 爬虫的介绍,和requests模块的简单使用

1.1 数据解析的三种方式。正则表达式, bs4, xpath

2.0 多任务(进程,协程,线程)爬虫:验证码识别,返回头储存,ip代理 介绍。 异步是什么,爬虫异步的方式。线程,进程,介绍

2.0.1 协程的 async/await 实现 爬虫 单线程 + 异步协程的实现

3.0 基于selenium 模块的 爬虫操作。 selenium 是一个用于Web应用程序测试的工具。Selenium测试直接运行在浏览器中,就像真正的用户在操作一样。

更多...

评论([[comments.sum]])

发表

加载更多([[item.son.length-2]])...

发表

2020-11 By chuan.

Python-flask & bootstrap-flask

图片外链来自:fghrsh

互联网ICP备案号:蜀ICP备2020031846号