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 Optionalfrom fastapi import FastAPIapp = 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 FastAPIapp = FastAPI()@app.get("/items/{item_id}")async def read_item(item_id):return {"item_id": item_id}

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

from fastapi import FastAPIapp = 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 Optionalfrom fastapi import FastAPI, Queryapp = 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, Queryapp = 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 Optionalfrom fastapi import Cookie, FastAPIapp = FastAPI()@app.get("/items/")
async def read_items(ads_id: Optional[str] = Cookie(None)):return {"ads_id": ads_id}

以及Header

from typing import Optionalfrom fastapi import FastAPI, Headerapp = FastAPI()@app.get("/items/")
async def read_items(user_agent: Optional[str] = Header(None)):return {"User-Agent": user_agent}

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

from typing import Optional, Setfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = Nonetags: 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, Setfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = Nonetags: 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, Setfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = Nonetags: 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 FastAPIapp = 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, Optionalfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = Nonetags: 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 FastAPIapp = FastAPI()@app.post("/items/", status_code=201)async def create_item(name: str):return {"name": name}

表单字段时,要使用Form

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

File用于定义客户端的上传文件(接收上传文件,要预先安装python-multipart):

from fastapi import FastAPI, File, UploadFileapp = 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, HTTPExceptionapp = 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, Setfrom fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = Nonetags: 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 Optionalfrom fastapi import FastAPIfrom fastapi.encoders import jsonable_encoderfrom pydantic import BaseModelfake_db = {}class Item(BaseModel):title: strtimestamp: datetimedescription: Optional[str] = Noneapp = 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 Optionalfrom fastapi import Depends, FastAPIapp = 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 Optionalfrom fastapi import Depends, FastAPIapp = 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 = qself.skip = skipself.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 Optionalfrom fastapi import Cookie, Depends, FastAPIapp = FastAPI()def query_extractor(q: Optional[str] = None):return qdef query_or_cookie_extractor(q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(None)
):if not q:return last_queryreturn 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, HTTPExceptionapp = 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, HTTPExceptionasync 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_keyapp = 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 Optionalfrom fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwtfrom passlib.context import CryptContextfrom pydantic import BaseModel# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30fake_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: strtoken_type: strclass TokenData(BaseModel):username: Optional[str] = Noneclass User(BaseModel):username: stremail: Optional[str] = Nonefull_name: Optional[str] = Nonedisabled: Optional[bool] = Noneclass UserInDB(User):hashed_password: strpwd_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 Falseif not verify_password(password, user.hashed_password):return Falsereturn userdef create_access_token(data: dict, expires_delta: Optional[timedelta] = None):to_encode = data.copy()if expires_delta:expire = datetime.utcnow() + expires_deltaelse:expire = datetime.utcnow() + timedelta(minutes=15)to_encode.update({"exp": expire})encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)return encoded_jwtasync 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_exceptiontoken_data = TokenData(username=username)except JWTError:raise credentials_exceptionuser = get_user(fake_users_db, username=token_data.username)if user is None:raise credentials_exceptionreturn userasync 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 timefrom fastapi import FastAPI, Requestapp = 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_timeresponse.headers["X-Process-Time"] = str(process_time)return response

十、跨域设置

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

from fastapi import FastAPIfrom fastapi.middleware.cors import CORSMiddlewareapp = 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 请求头列表。默认为 []。你可以使用 ['*'] 允许所有的请求头。AcceptAccept-LanguageContent-Language 以及 Content-Type 请求头总是允许 CORS 请求。
  • allow_credentials - 指示跨域请求支持 cookies。默认是 False。另外,允许凭证时 allow_origins 不能设定为 ['*'],必须指定源。
  • expose_headers - 指示可以被浏览器访问的响应头。默认为 []
  • max_age - 设定浏览器缓存 CORS 响应的最长时间,单位是秒。默认为 600

十一、APIRouter

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

from fastapi import APIRouterrouter = 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, HTTPExceptionfrom ..dependencies import get_token_headerrouter = 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, FastAPIapp = 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 FastAPIfrom fastapi.staticfiles import StaticFilesapp = FastAPI()app.mount("/static", StaticFiles(directory="static"), name="static")

十四、子应用

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

from fastapi import FastAPIapp = 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, Requestapp = 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 StaticFilesfrom fastapi.templating import Jinja2Templatesapp = 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, WebSocketfrom fastapi.responses import HTMLResponseapp = 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 FastAPIapp = 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]

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

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

C盘爆满,python pip无法安装应用

解决方法1 C盘扩容 从其他盘压缩空间&#xff0c;C盘使用压缩的空间进行扩容&#xff0c;治标不治本&#xff0c;以后C盘还会越来越大 解决方法2 转移pip安装目录 1. 获取显示pip安装目录 C:\Users\biewang>pip show pip Name: pip Version: 23.3.1 Summary: The PyPA r…

PHP基础知识和操作

PHP在线运行 https://c.runoob.com/compile/1/ https://www.sotool.net/php80 将驼峰字符串转化为蛇形字符串 <?phpfunction CamelToSnake($camelValue) {$initValue preg_replace(/\s/u, , $camelValue);$snakeValue strtolower(preg_replace(/(.)(?[A-Z])/u, &quo…

【实战技能】 单步运行源码分析,一期视频整明白FreeRTOS内核源码框架和运行机制,RTOS Trace链表功能展示

从源码的角度来看&#xff0c;OS内核源码就是通过各种链表组装起来的&#xff0c;FreeRTOS就是下面几个链表组成的。FreeRTOS的调度&#xff0c;任务切换就是倒腾这几个链表。而其它的几款OS是一个链表就一撸到底了&#xff0c;FreeRTOS是搞了好几个。所以视频里面就重点介绍下…

Docker容器资源限制 CPU / 内存 / 磁盘

在一台物理机上启动了多个docker容器时,就需要对内存及cpu做出相关的限制,以达到容器互不影响的目的。 1.限制容器对内存的使用 ⼀个 docker host 上会运⾏若⼲容器,每个容器都需要 CPU、内存和 IO 资源。对于 KVM,VMware 等虚拟化技 术,⽤户可以控制分配多少 CPU、内存…

代码随想录-刷题第十七天

110.平衡二叉树 题目链接&#xff1a;110. 平衡二叉树 思路&#xff1a;其实是判断左右子树的高度差是否大于1。判断高度的话仍然是采用后序遍历。 求深度可以从上到下去查&#xff0c;所以需要前序遍历&#xff08;中左右&#xff09;&#xff0c;而高度只能从下到上去查&a…

使用Redis做动态页面缓存,提高网页访问速度

目的 本关目的&#xff1a;实现使用Redis缓存网页。 相关知识 本文将教会你掌握&#xff1a;1&#xff0e;SETEX命令&#xff0c;2&#xff0e;hash()方法。 在动态生成网页的时候&#xff0c;通常会使用模板&#xff08;template&#xff09;来简化网页的生成&#xff0c;…

公募REITs交易规则详解

普通投资者是否可以参与基础设施公募REITs交易&#xff1f; 基础设施公募 REITs 采取封闭式运作,符合法定条件并经交易所依法审核同意后&#xff0c;可以上市交易。基础设施公募 REITs 的认购和交易实施适当性管理制度。尽管对普通投资者参与基础设施公募 REITs 的认购和交易没…

SpringBoot整合MongoDB

一、环境准备 1、添加 SpringData 依赖&#xff1a; <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-mongodb</artifactId> </dependency>2、配置 yml 文件 方式一&#xff1a; sprin…

新手安装Anaconda与Miniconda怎么选?

Python使用者的一大挑战是库的管理和环境的隔离&#xff0c; 常使用分布式管理系统&#xff0c;如Anaconda或Miniconda。接下来将探讨一下这两者的相似之处、差异和应用场景。 1、Anaconda和Miniconda的基本情况 Anaconda和Miniconda都是Continuum Analytics的开源项目&#x…

boost::throw_exception错误:修改VS代码生成异常选项为/EHsc

VS2013添加boost头文件和库文件路径后&#xff0c;代码编译报错&#xff1a; 错误 LNK2019 无法解析的外部符号 “void __cdecl boost::throw_exception(class std::exception const &)” (?throw_exceptionboostYAXAEBVexceptionstdZ)&#xff0c;该符号在函数 “public:…

uniapp打包的h5项目多了接口调用https://api.next.bspapp.com/client

产生跨域问题。 这个实际上是因为该项目在manifest.json文件中勾选了‘uni统计配置’导致的&#xff0c;取消勾选就可以了。 如果是小程序项目&#xff0c;在小程序开发者工具中添加可信任域名就可以了。 可以看看下面这个链接内容 uni-app H5跨域问题解决方案&#xff08;…

Python实现股票回测框架搭建,回测交易策略可行性。

文章目录 什么是回测框架?回测框架交易历史数据回测报告回测示例关于Python技术储备一、Python所有方向的学习路线二、Python基础学习视频三、精品Python学习书籍四、Python工具包项目源码合集①Python工具包②Python实战案例③Python小游戏源码五、面试资料六、Python兼职渠道…

建设纺织企业安全生产风险管控平台,降低生产风险

纺织行业是一个传统的行业&#xff0c;近年来面临着市场竞争加剧、成本压力上升、环保要求提高等挑战。纺织行业未来的发展需要紧跟市场需求和行业趋势&#xff0c;通过技术创新、消费升级、智能化生产、绿色环保和多元化市场等策略&#xff0c;提高产品附加值和市场竞争力&…

Footprint Analytics x Future3 万字研报:AI 与 Web3 数据行业融合的现状、竞争格局与未来机遇探析(上)

GPT的横空出世将全球的目光吸引至大语言模型&#xff0c;各行各业都尝试着利用这个“黑科技”提高工作效率&#xff0c;加速行业发展。Future3 Campus携手Footprint Analytics共同深入研究AI与Web3结合的无限可能&#xff0c;联合发布了《AI与Web3数据行业融合现状、竞争格局与…

写给初学者的 HarmonyOS 教程 -- 页面路由(router)

页面路由&#xff08;router&#xff09;是指在应用程序中实现不同页面之间的跳转和数据传递。 HarmonyOS 提供了 Router 模块&#xff0c;通过不同的 url 地址&#xff0c;可以方便地进行页面路由&#xff0c;轻松地访问不同的页面。 类似这样的效果&#xff1a; 页面跳转是…

香港服务器时间不准,差8小时

解决方案1 1、timedatectl查看系统时间 2、查看系统时区 ls /usr/share/zoneinfo 3、删除当前系统所处时区 rm /etc/localtime 4、创建软链接&#xff0c;以替换当前的时区信息 ln -s /usr/share/zoneinfo/Universal /etc/localtime 解决方案2 手动设置硬件时钟 1、设置系…

Hadoop学习笔记(HDP)-Part.15 安装HIVE

目录 Part.01 关于HDP Part.02 核心组件原理 Part.03 资源规划 Part.04 基础环境配置 Part.05 Yum源配置 Part.06 安装OracleJDK Part.07 安装MySQL Part.08 部署Ambari集群 Part.09 安装OpenLDAP Part.10 创建集群 Part.11 安装Kerberos Part.12 安装HDFS Part.13 安装Ranger …

【MODBUS】libmodbus库从Modbus从站读取值

libmodbus库提供了许多示例代码&#xff0c;用于演示如何使用该库进行Modbus通信。以下是一个简单的libmodbus库示例&#xff0c;用于从Modbus从站读取保持寄存器&#xff08;holding registers&#xff09;的值&#xff1a; #include <stdio.h> #include <stdlib.…

11.Java Spring

11.1Spring基础(核心概念,创建和使用,简单读取) 11.2SpringBoot 11.3SpringMVC 11.4MyBatis(基础) 11.5MyBatis(进阶) 11.6AOP 11.7统一功能处理 11.8事务 11.9密码加密,加盐算法(手动实现) 11.10Redis基础​​​​​​​