FastAPI 如何在FastAPI中的HTTP请求之间共享变量

在本文中，我们将介绍如何在FastAPI中实现在HTTP请求之间共享变量的方法。在Web应用程序中，有时需要在不同的HTTP请求之间共享某些变量，以便跟踪状态或存储全局数据。FastAPI提供了几种方法来实现这一目标。

使用全局变量

最简单的方式是使用全局变量来在HTTP请求之间共享数据。在FastAPI中，我们可以在主文件中定义一个全局变量，并在不同的请求处理函数中使用它。例如，假设我们想要跟踪应用程序中的总请求数量，我们可以创建一个名为request_counter的全局变量：

from fastapi import FastAPI

app = FastAPI()
request_counter = 0

@app.get("/")
async def root():
    global request_counter
    request_counter += 1
    return {"message": "Hello World", "request_counter": request_counter}

在上面的例子中，我们定义了一个名为request_counter的全局变量，并在根路由函数root()中对它进行递增操作。每次访问根路径时，我们将返回一个带有当前请求计数的JSON响应。

虽然使用全局变量可以在不同的请求之间共享数据，但是它可能存在线程安全性等问题，并且不适用于多进程应用程序。因此，我们还可以使用FastAPI提供的其他方法来实现更可靠和高效的数据共享机制。

使用依赖项

FastAPI提供了依赖项（Dependency）来实现在不同的请求之间共享数据。依赖项允许我们在请求处理函数中进行数据预处理、共享状态及资源管理等操作。通过使用依赖项，我们可以将跟踪状态和全局数据的任务从处理函数中分离出来，提高代码的可维护性和测试性。

我们可以定义一个依赖项函数来创建一个共享变量，并在需要访问该变量的请求处理函数中将其声明为参数。例如，假设我们想要在不同的请求中跟踪当前用户的信息，可以定义一个依赖项函数来获取当前用户的信息：

from fastapi import FastAPI, Depends

app = FastAPI()

def get_current_user():
    # 获取当前用户的逻辑
    return {"username": "JohnDoe"}

@app.get("/")
async def root(current_user: dict = Depends(get_current_user)):
    return {"message": "Hello, {}".format(current_user["username"])}

在上面的例子中，我们定义了一个名为get_current_user的依赖项函数，并在根路由函数中将其声明为参数。在每次请求根路径时，FastAPI会自动调用get_current_user函数获取当前用户的信息，并将其作为current_user参数传递给根路由函数。

使用依赖项的好处是它可以通过各种方式获取共享数据，例如从数据库、认证系统或缓存中获取数据。我们还可以使用依赖项来进行数据验证、异常处理和鉴权等操作，以及实现更复杂的应用逻辑。

使用全局上下文

FastAPI还提供了Request和BackgroundTasks等全局上下文对象，可以在不同的请求之间共享数据。Request对象用于访问当前请求的相关信息，例如路径参数、查询参数和请求主体等。BackgroundTasks对象用于在后台处理任务，例如发送异步邮件或执行定时任务。

我们可以通过在请求处理函数中声明Request或BackgroundTasks对象的参数来访问全局上下文。下面是一个示例，展示了如何在不同的请求中共享数据并使用全局上下文对象：

from fastapi import FastAPI, Request, BackgroundTasks

app = FastAPI()
request_counter = 0

@app.middleware("http")
async def add_request_counter(request: Request, call_next):
    global request_counter
    request_counter += 1
    response = await call_next(request)
    response.headers["X-Request-Counter"] = str(request_counter)
    return response

@app.get("/")
async def root(request: Request):
    return {"message": "Hello World", "request_counter": request.headers.get("X-Request-Counter")}

@app.post("/task")
async def create_task(background_tasks: BackgroundTasks):
    def _run_task():
        # 执行后台任务的逻辑
        pass

    background_tasks.add_task(_run_task)
    return {"message": "Task created"}

在上面的例子中，我们通过定义一个中间件函数add_request_counter来在每次请求时自动递增request_counter变量，并将其添加到响应的头部。在根路由函数root()中，我们可以通过request.headers来获取该头部并返回给客户端。另外，在/task路由中，我们使用BackgroundTasks对象来添加后台任务。