相关疑难解决方法(0)

我已经阅读了 FastAPI 关于中间件的文档（特别是中间件教程、CORS 中间件部分和高级中间件指南），但找不到如何编写可以使用该add_middleware函数添加的中间件类的具体示例（相反到使用装饰器添加的基本中间件功能），也不在此站点上。

我更喜欢使用add_middleware基于应用程序的装饰器的原因是，我想在共享库中编写一个中间件，该库将由多个不同的项目使用，因此我无法将其绑定到特定实例FastAPI。

所以我的问题是：你是如何做到的？

一直在尝试使用 FASTAPI 中间件获取请求的正文，但似乎我只能获取 request.headers 而不能获取正文。我需要主体才能获得用于检查数据库中某些内容的密钥。考虑中间件的日志记录或身份验证使用。这必须在不自定义 APIRouter 的情况下完成，只需要在中间件级别完成，以免影响应用程序的其余部分。

@app.middleware("http")
    async def TestCustomMiddleware(request: Request, call_next):
    print("Middleware works!", request.headers)

    response = await call_next(request)
    resp_body = [section async for section in response.__dict__['body_iterator']]
    print("BODY:", resp_body)
    return response

我能够得到这个，但有一个错误会破坏 POST 请求：

INFO:     Started server process [37160]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Middleware works! Headers({'content-type': 'application/json', 'user-agent': 'PostmanRuntime/7.26.8', 'accept': '*/*', 'cache-control': 'no-cache', 'postman-token': 'ca6839ec-833d-45c0-9b52-8f904db13966', 'host': 'localhost:8000', 'accept-encoding': 'gzip, deflate, br', 'connection': 'keep-alive', 'content-length': '12'}) …

有没有办法在中间件中获取响应内容？以下代码是从此处复制的。

@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

HTTP如何在 FastAPI 的请求之间共享变量值？例如，我有一个POST请求，其中我获取一些音频文件，然后将它们的信息转换为 Pandas Dataframe。我想在请求Dataframe中发送它GET，但我无法访问Dataframe请求GET范围。

@app.post(
    path="/upload-audios/",
    status_code=status.HTTP_200_OK
)
async def upload_audios(audios: list[UploadFile] = File(...)):
    filenames = [audio.filename for audio in audios]
    audio_data = [audio.file for audio in audios]
    new_data = []
    final_data = []
    header = ["name", "file"]
    for i in range(len(audios)):
        new_data = [filenames[i], audio_data[i]]
        final_data.append(new_data)
    new_df = pd.DataFrame(final_data, columns=header)
    return f"You have uploaded {len(audios)} audios which names are: {filenames}"

@app.get("/get-dataframe/")
async def get_dataframe():
    pass

我正在尝试将一个大文件 (\xe2\x89\xa53GB) 上传到我的 FastAPI 服务器，而不将整个文件加载到内存中，因为我的服务器只有 2GB 可用内存。

服务器端：

async def uploadfiles(upload_file: UploadFile = File(...):\n

客户端：

m = MultipartEncoder(fields = {"upload_file":open(file_name,\'rb\')})\nprefix = "http://xxx:5000"\nurl = "{}/v1/uploadfiles".format(prefix)\ntry:\n    req = requests.post(\n    url,\n    data=m,\n    verify=False,\n            )\n

返回：

async def uploadfiles(upload_file: UploadFile = File(...):\n

我不确定MultipartEncoder实际发送到服务器的内容，因此请求不匹配。有任何想法吗？

我有一个 FastAPI 应用程序，我希望将默认日志写入 STDOUT，并使用 JSON 格式的以下数据：

应用程序日志应如下所示：

{
 "XYZ": {
   "log": {
     "level": "info",
     "type": "app",
     "timestamp": "2022-01-16T08:30:08.181Z",
     "file": "api/predictor/predict.py",
     "line": 34,
     "threadId": 435454,
     "message": "API Server started on port 8080 (development)"
    }
  }
}

访问日志应如下所示：

{
 "XYZ": {
   "log": {
     "level": "info",
     "type": "access",
     "timestamp": "2022-01-16T08:30:08.181Z",
     "message": "GET /app/health 200 6ms"
   },
   "req": {
     "url": "/app/health",
     "headers": {
       "host": "localhost:8080",
       "user-agent": "curl/7.68.0",
       "accept": "*/*"
     },
     "method": "GET",
     "httpVersion": "1.1",
     "originalUrl": "/app/health",
     "query": {}
   },
   "res": {
     "statusCode": 200,
     "body": …

当 FastAPI 端点发生异常时，我尝试使用后台任务生成日志：

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

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

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    if "hello" in email:
        background_tasks.add_task(write_notification, message="helloworld")
        raise HTTPException(status_code=500, detail="example error")

    background_tasks.add_task(write_notification, message="hello world.")
    return {"message": "Notification sent in the background"}

但是，不会生成日志，因为根据此处和此处的文档，后台任务“仅”在return执行语句后运行。

有什么解决方法吗？

我有这样的中间件

class RequestContext(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
        request_id = request_ctx.set(str(uuid4()))  # generate uuid to request
        body = await request.body()
        if body:
            logger.info(...)  # log request with body
        else:
            logger.info(...)  # log request without body
 
        response = await call_next(request)
        response.headers['X-Request-ID'] = request_ctx.get()
        logger.info("%s" % (response.status_code))
        request_ctx.reset(request_id)

        return response

因此该行body = await request.body()冻结了所有具有正文的请求，而我从所有请求中获得了 504 个请求。在这种情况下如何安全地读取请求正文？我只想记录请求参数。

我有一个GET在路径中带有请求参数的方法：

@router.get('/users/{user_id}')
async def get_user_from_string(user_id: str):
    return User(user_id)

'/users/{user_id}'是否可以从请求中获取基本 url 原始路径（即）？

我尝试使用以下方式：

path = [route for route in request.scope['router'].routes if
        route.endpoint == request.scope['endpoint']][0].path

但这不起作用，我得到：

属性错误：“安装”对象没有属性“端点”

我有一个 FastAPIGET端点，它返回大量 JSON 数据（约 160,000 行和 45 列）。毫不奇怪，使用返回数据非常json.dumps()慢。我首先使用文件中的数据读取数据json.loads()，并根据输入的参数对其进行过滤。有没有比使用更快的方法将数据返回给用户return data？以目前的状态，需要将近一分钟的时间。

我的代码目前如下所示：

# helper function to parse parquet file (where data is stored)
def parse_parquet(file_path):
    df = pd.read_parquet(file_path)
    result = df.to_json(orient = 'records')
    parsed = json.loads(result)
    return parsed
    

@app.get('/endpoint')
# has several more parameters
async def some_function(year = int | None = None, id = str | None = None):
    if year is None:
        data = parse_parquet(f'path/{year}_data.parquet')
    # no year
    if year is …