当前位置：首页 > news >正文

同步和异步路由

news 2024/10/21 11:24:40

FastAPI 中同时使用同步和异步路由。解析并解释其中的关键点。

代码解析

from fastapi import FastAPI
import threading
import time
import asyncio
app = FastAPI(routes=None)@app.get(path="/async")
async def asyncdef():await asyncio.sleep(10)print("当前协程运行的线程ID:", threading.current_thread().ident)return {"index": "async"}@app.get(path="/sync")
def syncdef():time.sleep(10)print("当前普通函数运行的线程ID:",threading.current_thread().ident)return {"index": "sync"}if __name__ == "__main__":import uvicornimport osapp_modeel_name = os.path.basename(__file__).replace(".py", "")print(app_modeel_name)uvicorn.run(f"{app_modeel_name}:app", host='0.0.0.0', reload=True)

导入模块

from fastapi import FastAPI
import threading
import time
import asyncio

from fastapi import FastAPI 导入了 FastAPI 框架。
import threading 和 import time 分别导入了线程和时间模块。
import asyncio 导入了异步编程模块。

创建 FastAPI 应用

app = FastAPI(routes=None)

创建了一个 FastAPI 应用实例 app。routes=None 表示没有预先定义的路由。

定义异步路由

@app.get(path="/async")
async def asyncdef():await asyncio.sleep(10)print("当前协程运行的线程ID:", threading.current_thread().ident)return {"index": "async"}

@app.get(path="/async") 是一个装饰器，将 asyncdef 函数注册为处理 /async 路径的 GET 请求。
async def asyncdef(): 定义了一个异步函数 asyncdef。
await asyncio.sleep(10) 使当前协程暂停执行10秒，同时释放控制权给事件循环。
print("当前协程运行的线程ID:", threading.current_thread().ident) 打印当前协程运行的线程ID。
return {"index": "async"} 返回一个 JSON 响应。

定义同步路由

@app.get(path="/sync")
def syncdef():time.sleep(10)print("当前普通函数运行的线程ID:",threading.current_thread().ident)return {"index": "sync"}

@app.get(path="/sync") 是一个装饰器，将 syncdef 函数注册为处理 /sync 路径的 GET 请求。
def syncdef(): 定义了一个同步函数 syncdef。
time.sleep(10) 使当前线程暂停执行10秒。
print("当前普通函数运行的线程ID:",threading.current_thread().ident) 打印当前同步函数运行的线程ID。
return {"index": "sync"} 返回一个 JSON 响应。

启动应用

if __name__ == "__main__":import uvicornimport osapp_modeel_name = os.path.basename(__file__).replace(".py", "")print(app_modeel_name)uvicorn.run(f"{app_modeel_name}:app", host='0.0.0.0', reload=True)

if __name__ == "__main__": 确保以下代码仅在直接运行此脚本时执行。
import uvicorn 导入了 Uvicorn，这是一个用于运行 FastAPI 应用的 ASGI 服务器。
import os 导入了操作系统模块。
app_modeel_name = os.path.basename(__file__).replace(".py", "") 获取当前脚本文件名（不包括扩展名）。
uvicorn.run(f"{app_modeel_name}:app", host='0.0.0.0', reload=True) 启动 Uvicorn 服务器，监听所有网络接口（host='0.0.0.0'），并启用自动重载（reload=True）。