Python 高性能 web 框架 - FastApi 全面指南

原文：Python 高性能 web 框架 - FastApi 全面指南 - 知乎

一、简介

FastAPI 是一个用于构建 API 的现代、快速（高性能）的 web 框架，使用 Python 3.6+ 并基于标准的 Python 类型提示。

它具有如下这些优点：

  • 快速：可与 NodeJS 和 Go 比肩的极高性能（归功于 Starlette 和 Pydantic）
  • 高效编码：提高功能开发速度约 200％ 至 300％
  • 更少 bug：减少约 40％ 的人为（开发者）导致错误。
  • 智能：极佳的编辑器支持。处处皆可自动补全，减少调试时间
  • 简单：设计的易于使用和学习，阅读文档的时间更短
  • 简短：使代码重复最小化。通过不同的参数声明实现丰富功能。bug 更少
  • 健壮：生产可用级别的代码。还有自动生成的交互式文档
  • 标准化：基于（并完全兼容）API 的相关开放标准：OpenAPI (以前被称为 Swagger) 和 JSON Schema。

二、安装

pip install fastapi

ASGI 服务器可以使用uvicorn：

pip install uvicorn[standard]

三、简单示例

创建一个 main.py 文件并写入以下内容:

from typing import Optional

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):
    return {"item_id": item_id, "q": q}

启动服务器：

uvicorn main:app --reload

访问URL：http://127.0.0.1:8000/items/5?q=somequery，你将会看到如下 JSON 响应：

{"item_id": 5, "q": "somequery"}

访问URL：http://127.0.0.1:8000/docs，你会看到自动生成的交互式 API 文档，由Swagger UI 生成：

访问URL：http://127.0.0.1:8000/redoc，你会看到另一个自动生成的文档（由ReDoc生成）：

四、请求

使用与 Python 格式化字符串相同的语法来声明路径"参数"或"变量"：

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/{item_id}")

async def read_item(item_id):

    return {"item_id": item_id}

路径参数item_id的值将作为参数item_id传递给你的函数。声明不属于路径参数的其他函数参数时，它们将被自动解释为"查询字符串"参数：

from fastapi import FastAPI

app = FastAPI()

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


@app.get("/items/")

async def read_item(skip: int = 0, limit: int = 10):

    return fake_items_db[skip : skip + limit]

查询字符串是键值对的集合，这些键值对位于 URL 的？之后，并以&符号分隔。

可以使用Query对查询进行额外的校验：

from typing import Optional


from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Optional[str] = Query(None, max_length=50)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results

Query 有如下这些字段校验：

  • min_length 最小长度
  • max_length 最大长度
  • regex 正则匹配
  • Query 第一个参数为默认值，...表示是必需的

Path和Query用法一样，也能对查询字段进行校验。

而且你还可以声明数值校验：

from fastapi import FastAPI, Path, Query

app = FastAPI()


@app.get("/items/{item_id}")
async def read_items(
    *,
    item_id: int = Path(..., title="The ID of the item to get", ge=0, le=1000),
    q: str,

    size: float = Query(..., gt=0, lt=10.5)

):
    results = {"item_id": item_id}
    if q:
        results.update({"q": q})
    return results
  • gt：大于
  • ge：大于等于
  • lt：小于
  • le：小于等于

类似的还有Cookie：

from typing import Optional


from fastapi import Cookie, FastAPI


app = FastAPI()


@app.get("/items/")
async def read_items(ads_id: Optional[str] = Cookie(None)):
    return {"ads_id": ads_id}

以及Header：

from typing import Optional


from fastapi import FastAPI, Header


app = FastAPI()


@app.get("/items/")
async def read_items(user_agent: Optional[str] = Header(None)):
    return {"User-Agent": user_agent}

还可以为路径设置tags标签进行分组：

from typing import Optional, Set

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []



@app.post("/items/", response_model=Item, tags=["items"])

async def create_item(item: Item):
    return item



@app.get("/items/", tags=["items"])

async def read_items():
    return [{"name": "Foo", "price": 42}]



@app.get("/users/", tags=["users"])

async def read_users():
    return [{"username": "johndoe"}]

还可以设置summary 和 description:

from typing import Optional, Set

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []


@app.post(
    "/items/",
    response_model=Item,

    summary="Create an item",

    description="Create an item with all the information, name, description, price, tax and a set of unique tags",

)
async def create_item(item: Item):
    return item

多行注释：

from typing import Optional, Set

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []


@app.post("/items/", response_model=Item, summary="Create an item")
async def create_item(item: Item):

    """

    Create an item with all the information:



    - **name**: each item must have a name

    - **description**: a long description

    - **price**: required

    - **tax**: if the item doesn't have tax, you can omit this

    - **tags**: a set of unique tag strings for this item

    """

    return item

废弃路由：

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/", tags=["items"])
async def read_items():
    return [{"name": "Foo", "price": 42}]


@app.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "johndoe"}]



@app.get("/elements/", tags=["items"], deprecated=True)

async def read_elements():
    return [{"item_id": "Foo"}]

五、响应

使用response_model参数来声明用于响应的模型：

from typing import List, Optional

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: List[str] = []



@app.post("/items/", response_model=Item)

async def create_item(item: Item):
    return item
  • response_model_exclude_unset=True：响应中将不会包含那些默认值，而是仅有实际设置的值
  • response_model_include包含哪些属性
  • response_model_exclude省略某些属性

status_code参数来声明用于响应的 HTTP 状态码：

from fastapi import FastAPI

app = FastAPI()



@app.post("/items/", status_code=201)

async def create_item(name: str):
    return {"name": name}

表单字段时，要使用Form：

from fastapi import FastAPI, Form


app = FastAPI()


@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    return {"username": username}

File用于定义客户端的上传文件（接收上传文件，要预先安装python-multipart）：

from fastapi import FastAPI, File, UploadFile


app = 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 = File(...)):
    return {"filename": file.filename}

向客户端返回 HTTP 错误响应，可以使用HTTPException。

from fastapi import FastAPI, HTTPException


app = FastAPI()

items = {"foo": "The Foo Wrestlers"}


@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

使用response_description设置响应描述：

from typing import Optional, Set

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []


@app.post(
    "/items/",
    response_model=Item,
    summary="Create an item",

    response_description="The created item",

)
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    """
    return item

六、JSON兼容

在某些情况下，你可能需要把数据（例如Pydantic模型）转换成JSON形式，例如存储在数据库中，这时候你就需要用到jsonable_encoder()方法。

from datetime import datetime
from typing import Optional

from fastapi import FastAPI

from fastapi.encoders import jsonable_encoder

from pydantic import BaseModel

fake_db = {}


class Item(BaseModel):
    title: str
    timestamp: datetime
    description: Optional[str] = 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

七、依赖注入

FastAPI 提供了简单易用，但功能强大的依赖注入系统，可以让开发人员轻松地把组件集成至FastAPI。

什么是「依赖注入」？

依赖注入是一种消除类之间依赖关系的设计模式。把有依赖关系的类放到容器中，解析出这些类的实例，就是依赖注入。目的是实现类的解耦。

示例：

from typing import Optional

from fastapi import Depends, FastAPI

app = FastAPI()



async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):

    return {"q": q, "skip": skip, "limit": limit}



@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons

本例中的依赖项预期接收如下参数：

  • 类型为 str 的可选查询参数 q
  • 类型为 int 的可选查询参数 skip，默认值是 0
  • 类型为 int 的可选查询参数 limit，默认值是 100

然后，依赖项函数返回包含这些值的 dict。

使用Class作为依赖：

from typing import Optional

from fastapi import Depends, FastAPI

app = FastAPI()


fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


@app.get("/items/")

async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):

    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response

使用嵌套子依赖：

from typing import Optional

from fastapi import Cookie, Depends, FastAPI

app = FastAPI()


def query_extractor(q: Optional[str] = None):
    return q


def query_or_cookie_extractor(
    q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(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}

在路径中使用依赖：

from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()


async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key



@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])

async def read_items():
    return [{"item": "Foo"}, {"item": "Bar"}]

全局依赖项，可以为所有路径操作应用该依赖项：

from fastapi import Depends, FastAPI, Header, HTTPException


async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token header invalid")


async def verify_key(x_key: str = Header(...)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key header invalid")
    return x_key



app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])



@app.get("/items/")
async def read_items():
    return [{"item": "Portal Gun"}, {"item": "Plumbus"}]


@app.get("/users/")
async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]

八、安全

在许多框架和系统中，仅处理安全性和身份认证就会花费大量的精力和代码（在许多情况下，可能占编写的所有代码的 50％ 或更多）。

FastAPI 提供了多种工具，可帮助你以标准的方式轻松、快速地处理安全性，而无需研究和学习所有的安全规范。

JWT 表示 「JSON Web Tokens」。

它是一个将 JSON 对象编码为密集且没有空格的长字符串的标准。字符串看起来像这样：

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c

它没有被加密，因此任何人都可以从字符串内容中还原数据。

但它经过了签名。因此，当你收到一个由你发出的令牌时，可以校验令牌是否真的由你发出。

通过这种方式，你可以创建一个有效期为 1 周的令牌。然后当用户第二天使用令牌重新访问时，你知道该用户仍然处于登入状态。

一周后令牌将会过期，用户将不会通过认证，必须再次登录才能获得一个新令牌。而且如果用户（或第三方）试图修改令牌以篡改过期时间，你将因为签名不匹配而能够发觉。

OAuth2

OAuth2是一个规范，它定义了几种处理身份认证和授权的方法。

它是一个相当广泛的规范，涵盖了一些复杂的使用场景。

它包括了使用「第三方」进行身份认证的方法。

这就是所有带有「使用 Facebook，Google，Twitter，GitHub 登录」的系统背后所使用的机制。

下面演示了如何使用OAuth2 和 JWT进行用户验证。

from datetime import datetime, timedelta
from typing import Optional

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt

from passlib.context import CryptContext

from pydantic import BaseModel

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    }
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: Optional[str] = None


class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None


class UserInDB(User):
    hashed_password: str



pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()



def verify_password(plain_password, hashed_password):

    return pwd_context.verify(plain_password, hashed_password)




def get_password_hash(password):
    return pwd_context.hash(password)



def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)



def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)

    if not user:

        return False

    if not verify_password(password, user.hashed_password):

        return False

    return user



def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(current_user: User = Depends(get_current_active_user)):
    return [{"item_id": "Foo", "owner": current_user.username}]
  • OAuth2PasswordBearer：访问tokenUrl地址，获取token并返回
  • OAuth2PasswordRequestForm是一个类依赖项，声明了如下的请求表单：
    • username
    • password
    • 一个可选的 scope 字段，是一个由空格分隔的字符串组成的大字符串
    • 一个可选的 grant_type
    • 一个可选的 client_id
    • 一个可选的 client_secret

九、中间件

"中间件"是一个函数,它在每个请求被特定的路径操作处理之前,以及在每个响应返回之前工作。

要创建中间件你可以在函数的顶部使用装饰器 @app.middleware("http").

中间件参数接收如下参数:

  • request
  • 一个函数 call_next 它将接收 request 作为参数
    • 这个函数将 request 传递给相应的 路径操作
    • 然后它将返回由相应的路径操作生成的 response
  • 然后你可以在返回 response 前进一步修改它
import time

from fastapi import FastAPI, Request

app = FastAPI()



@app.middleware("http")

async def add_process_time_header(request: Request, call_next):

    start_time = time.time()

    response = await call_next(request)

    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)

    return response

十、跨域设置

你可以在FastAPI应用中使用CORSMiddleware来配置跨域：

from fastapi import FastAPI

from fastapi.middleware.cors import CORSMiddleware


app = FastAPI()


origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",

]



app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],

)



@app.get("/")
async def main():
    return {"message": "Hello World"}

它支持以下参数：

  • allow_origins - 一个允许跨域请求的源列表。例如 ['https://example.org', 'https://www.example.org']。你可以使用 ['*'] 允许任何源。
  • allow_origin_regex - 一个正则表达式字符串，匹配的源允许跨域请求。例如 'https://.*\.example\.org'。
  • allow_methods - 一个允许跨域请求的 HTTP 方法列表。默认为 ['GET']。你可以使用 ['*'] 来允许所有标准方法。
  • allow_headers - 一个允许跨域请求的 HTTP 请求头列表。默认为 []。你可以使用 ['*'] 允许所有的请求头。Accept、Accept-Language、Content-Language 以及 Content-Type 请求头总是允许 CORS 请求。
  • allow_credentials - 指示跨域请求支持 cookies。默认是 False。另外，允许凭证时 allow_origins 不能设定为 ['*']，必须指定源。
  • expose_headers - 指示可以被浏览器访问的响应头。默认为 []。
  • max_age - 设定浏览器缓存 CORS 响应的最长时间，单位是秒。默认为 600。

十一、APIRouter

使用APIRouter同样也能对路由进行操作：

from fastapi import APIRouter

router = APIRouter()



@router.get("/users/", tags=["users"])

async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]



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

为所有路径进行同样的操作：

from fastapi import APIRouter, Depends, HTTPException

from ..dependencies import get_token_header


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



fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}



@router.get("/")

async def read_items():
    return fake_items_db



@router.get("/{item_id}")

async def read_item(item_id: str):
    if item_id not in fake_items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"name": fake_items_db[item_id]["name"], "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 != "plumbus":
        raise HTTPException(
            status_code=403, detail="You can only update the item: plumbus"
        )
    return {"item_id": item_id, "name": "The great Plumbus"}

该示例，就为所有的路径添加了前缀，标签、依赖和返回，而不用在每个路径上单独声明，简化了代码。

十二、Background Tasks

background tasks 就是在返回响应之后立即运行的任务。

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

十三、静态文件

首先需要安装aiofiles：

pip install aiofiles

使用：

from fastapi import FastAPI

from fastapi.staticfiles import StaticFiles


app = FastAPI()


app.mount("/static", StaticFiles(directory="static"), name="static")

十四、子应用

如果你有2个独立的FastAPI的应用，你可以设置一个为主应用，另外一个为子应用：

from fastapi import FastAPI


app = FastAPI()




@app.get("/app")

def read_main():
    return {"message": "Hello World from main app"}



subapi = FastAPI()


@subapi.get("/sub")
def read_sub():
    return {"message": "Hello World from sub API"}


app.mount("/subapi", subapi)

十五、代理

可以使用root_path来设置代理。

使用命令行：

uvicorn main:app --root-path /api/v1

或者在代码中设置：

from fastapi import FastAPI, Request


app = FastAPI(root_path="/api/v1")



@app.get("/app")
def read_main(request: Request):
    return {"message": "Hello World", "root_path": request.scope.get("root_path")}

十六、使用模板

你可以在FastAPI中使用任何模板，常用的选择是Jinja2。

pip install jinja2

使用：

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles

from fastapi.templating import Jinja2Templates


app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")



templates = Jinja2Templates(directory="templates")



@app.get("/items/{id}", response_class=HTMLResponse)

async def read_item(request: Request, id: str):
    return templates.TemplateResponse("item.html", {"request": request, "id": id})

模板文件templates/item.html：

<html>
<head>
    <title>Item Details</title>
    <link href="{{ url_for('static', path='/styles.css') }}" rel="stylesheet">
</head>
<body>

    <h1>Item ID: {{ id }}</h1>

</body>
</html>

十七：WebSockets

from fastapi import FastAPI, WebSocket

from fastapi.responses import HTMLResponse

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>Send</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:8000/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                var content = document.createTextNode(event.data)
                message.appendChild(content)
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""


@app.get("/")
async def get():
    return HTMLResponse(html)



@app.websocket("/ws")

async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

十八、startup - shutdown事件

你可以设置应用的启动和关闭事件回调函数：

from fastapi import FastAPI

app = FastAPI()

items = {}

@app.on_event("shutdown")
def shutdown_event():
    with open("log.txt", mode="a") as log:
        log.write("Application shutdown")


@app.on_event("startup")
async def startup_event():
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}


@app.get("/items/{item_id}")
async def read_items(item_id: str):
    return items[item_id]

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/210882.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

shell编程系列(11)-使用grep查找文本

shell编程系列(11)-使用grep查找文本

文章目录 前言grep的使用根据关键字查找反向查找 结语 前言 grep命令也是我们在日常使用linux&#xff0c;编写shell脚本中会用到的一个高频命令&#xff0c;grep主要是帮助我们查找我们想要的内容&#xff0c;类似于我们在office word里面的 Ctrl f 查找功能&#xff0c;但是…
阅读更多...
LabVIEW在不同操作系统上使VI、可执行文件或安装程序

LabVIEW在不同操作系统上使VI、可执行文件或安装程序

LabVIEW在不同操作系统上使VI、可执行文件或安装程序 LabVIEW可以在多个操作系统上运行&#xff0c;主要支持以下几种操作系统&#xff1a; Windows&#xff1a; LabVIEW在各个版本的Windows操作系统上都能运行&#xff0c;包括Windows 7、Windows 8和Windows10。LabVIEW为Wi…
阅读更多...
docker容器中创建非root用户

docker容器中创建非root用户

简介 用 docker 也有一段时间了&#xff0c;一直在 docker 容器中使用 root 用户肆意操作。直到部署 stable diffusion webui 我才发现无法使用 root 用户运行它&#xff0c;于是才幡然醒悟&#xff1a;是时候搞个非 root 用户了。 我使用的 docker 镜像文件是 centos:centos…
阅读更多...
使用系统ProgressBar实现三色进度条

使用系统ProgressBar实现三色进度条

使用系统ProgressBar实现如图三色进度条&#xff1a; //布局中<ProgressBarandroid:layout_width"0dp"android:layout_height"8dp"android:layout_marginLeft"16dp"app:layout_constraintBottom_toBottomOf"id/photo"app:layout_c…
阅读更多...
解决报错：error: (-215:Assertion failed) inv_scale_x ＞ 0 in function ‘cv::resize‘

解决报错：error: (-215:Assertion failed) inv_scale_x ＞ 0 in function ‘cv::resize‘

需求背景 欲使用opencv的resize函数将图像沿着纵轴放大一倍&#xff0c;即原来的图像大小为(384, 512), 现在需要将图像放大为(768, 512)。 源码 import cv2 import numpy as np# 生成初始图像 img np.zeros((384, 512), dtypenp.uint8) img[172:212, 32:-32] 255 H, W …
阅读更多...
Python爬虫教程27：秀啊！用Pandas 也能爬虫？？

Python爬虫教程27：秀啊！用Pandas 也能爬虫？？

说到爬虫&#xff0c;大家可能都知道requests、re、scrapy、selenium等等一些工具库。虽然它低调&#xff0c;但功能非常强大&#xff0c;用于抓取Table表格型数据时&#xff0c;简直是个神器&#xff0c;没有必要去F12研究HTML页面结构甚至写正则表达式解析字段。 #我的Pytho…
阅读更多...
Python教程78：聊聊exec和eval()函数，有什么用法区别

Python教程78：聊聊exec和eval()函数，有什么用法区别

exec 和 eval 是 Python 中的两个内置函数&#xff0c;它们都可以执行Python代码&#xff0c;但它们的使用方式和目的有所不同。 1.exec()函数用于执行动态的 Python 代码&#xff0c;你可以使用exec来执行存储在字符串或对象代码中的 Python 代码。exec 不会返回任何结果&…
阅读更多...
Python遥感开发之批量镶嵌

Python遥感开发之批量镶嵌

Python遥感开发之批量镶嵌 1.ArcGis镶嵌2.Arcpy实现镶嵌2.1 Arcpy实现单个镶嵌2.2 Arcpy实现批量镶嵌 3.GDAL实现镶嵌 前言&#xff1a;主要介绍了遥感数据的镶嵌&#xff0c;其中包括使用ArcGis如何完成镶嵌&#xff0c;如何使用Arcpy和GDAL完成镶嵌。 1.ArcGis镶嵌 是ArcGis…
阅读更多...
记一次若依二开的简单流程

记一次若依二开的简单流程

记一次若依二开的简单流程 前言: 搞Java后端的应该都知道若依框架&#xff0c;是一个十分强大且功能齐全的开源的快速开发平台&#xff0c;且毫无保留给个人及企业免费使用。很多中小型公司会直接在该系统上进行二次开发使用。本文记录一次使用若依二开零编码的简单实现&#…
阅读更多...
Fiddler抓包工具之fiddler设置断点和简单的并发测试

Fiddler抓包工具之fiddler设置断点和简单的并发测试

断点有两种方式&#xff1a; 1、全局断点 2、局部断点 全局断点 全局断点的特点是&#xff1a;不能针对一个请求&#xff0c;是给所有抓到的请求打断点 全局断点如何设置&#xff1a; 1、快速设置断点&#xff1a;直接点击底部状态栏断点处 &#xff1b;点击第一下是请求…
阅读更多...
vcruntime140.dll无法继续执行代码五种解决方法修复教程

vcruntime140.dll无法继续执行代码五种解决方法修复教程

在电脑使用过程中&#xff0c;我们可能会遇到一些常见的错误提示&#xff0c;其中之一就是“vcruntime140.dll丢失”。这个错误通常会导致某些应用程序无法正常运行。本文将介绍vcruntime140.dll丢失对电脑的影响以及如何修复这个问题&#xff0c;并提供一些预防措施&#xff0…
阅读更多...
Ubuntu系统CLion安装与Ubuntu下菜单启动图标设置

Ubuntu系统CLion安装与Ubuntu下菜单启动图标设置

Ubuntu系统CLion安装 pycharm 同理。 参考官网安装过程&#xff1a;官网安装过程 下载linux tar.gz包 # 解压 sudo tar -xzvf CLion-*.tar.gz -C /opt/ sh /opt/clion-*/bin/clion.sh其中第二个命令是启动CLion命令 clion安装完以后&#xff0c;不会在桌面或者菜单栏建立图…
阅读更多...
半监督语义分割综述

半监督语义分割综述

paper link&#xff1a;https://arxiv.org/pdf/2302.09899.pdf 1. Introduction 图像分割是最古老、研究最广泛的计算机视觉 (CV) 问题之一。图像分割是指将图像划分为不同的非重叠区域&#xff0c;并将相应的标签分配给图像中的每个像素&#xff0c;最终获得ROI区域位置及其类…
阅读更多...
Linux - 动态库的加载 和 重谈进程地址空间 - vscode 当中的 Remote - SSH 插件

Linux - 动态库的加载 和 重谈进程地址空间 - vscode 当中的 Remote - SSH 插件

推书&#xff1a;《现代操作系统》《操作系统--精髓于设计原理》《UNIX环境高级编程》 目录 前言 程序的加载 程序没有加载之前的地址&#xff08;此时还是程序&#xff09; 程序被加载到内存之后&#xff08;此时是进程&#xff09; 动态库的地址 静态库的不加载&#xff…
阅读更多...
海外之路，英语相助：我的雅思考试之旅（成功以此申请到了港科大硕士）

海外之路，英语相助：我的雅思考试之旅（成功以此申请到了港科大硕士）

文章目录 背景注册过程 雅思考试之旅全程记录2023年2月14日 - 报名补充2023年2月15日 - 模拟考试2023年2月16日 - 在线考试失利2023年2月20日 - 线下考试再战成功2023年2月23日 - 成绩出来了 考试经验寄送成绩问题请求额外的IELTS成绩寄送 结语 背景 在新加坡Shopee工作了两年…
阅读更多...
Linux 内核源码各版本下载

Linux 内核源码各版本下载

下载地址&#xff1a; kernel/git/stable/linux.git - Linux kernel stable treehttps://git.kernel.org/pub/scm/linux/kernel/git/stable/linux.git/refs/ 1. Linux 内核的基本概念&#xff1a; 内核是什么&#xff1f; 内核是操作系统的核心部分&#xff0c;负责管理系统…
阅读更多...
Active Stereo Without Pattern Projector论文精读

Active Stereo Without Pattern Projector论文精读

1.背景补充 主动立体相机和被动立体相机的主要区别在于它们获取立体视觉信息的方式 主动立体相机12&#xff1a; 主动立体视觉是指寻找最佳的视角去重建目标或者场景1。主动视觉的实现方式通常有&#xff1a;改变环境中的光照条件、改变相机的视角、移动相机自身位置等&…
阅读更多...
P1 嵌入式开发之什么是Linux应用开发

P1 嵌入式开发之什么是Linux应用开发

目录 前言 01 .Linux应用与裸机编程、驱动编程之间的区别 1.1裸机编程&#xff1a; 1.2 驱动编程 1.3应用编程 前言 &#x1f3ac; 个人主页&#xff1a;ChenPi &#x1f43b;推荐专栏1: 《C_ChenPi的博客-CSDN博客》✨✨✨ &#x1f525; 推荐专栏2: 《Linux C应用编程&a…
阅读更多...
蓝桥杯物联网竞赛_STM32L071_7_LORA发送接收

蓝桥杯物联网竞赛_STM32L071_7_LORA发送接收

理论&#xff1a; LoRa&#xff08;长距离低功耗无线射频&#xff09;是利用无线电频谱中的电磁波进行通信的一种技术。它使用无线电频段中的射频信号来传输数据。电磁波是一种波动形式的能量&#xff0c;具有特定的频率和波长&#xff0c;可以传播在真空和大多数介质中&#x…
阅读更多...
腾讯云云服务器功能与优势

腾讯云云服务器功能与优势

腾讯云云服务器&#xff08;Cloud Virtual Machine&#xff0c;CVM&#xff09;是腾讯云提供的可扩展的计算服务。使用云服务器 CVM 避免了使用传统服务器时需要预估资源用量及前期投入的问题&#xff0c;帮助您在短时间内快速启动任意数量的云服务器并及时部署应用程序。 云服…
阅读更多...
最新文章

Copyright @ 2022~2023