相关疑难解决方法(0)

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

@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 编写的 python Web 应用程序（通过 uvicorn 运行）。在我的应用程序中，我使用的是标准logging模块，该模块使用TimedRotatingFileHandler. 由于我正在登录文件，因此我担心性能。假设我在一个函数中有 3 条日志消息，每次调用端点时都会调用该函数/test。现在想象 1000 个客户端/test同时请求 - 即向日志文件写入 3000 次。

我担心这会影响应用程序的性能，因为 IO 任务非常耗时。我应该使用某种形式的异步日志记录吗？也许打开一个新线程来写入文件？我尝试用谷歌搜索答案，但没有找到标准化的方法。那么它还需要吗？如果是，我应该如何处理这个问题？谢谢！

注意：这个问题与这里的问题不同，因为我需要它与 Swagger 一起工作。

给定一个 FastAPIGET端点，我希望允许任意一组 URL 参数，同时保持 Swagger 支持。

我的用例是我想要支持一组类似于 JSON API 的查询参数，如下所示：

/api/books/?include=author&sort=name,zip&sort[author]=-lname&fields=name,phone,street

使用方括号使我无法使用传统的类来对查询参数进行建模，因此我直接使用该Request对象。但是，我想使用 Swagger 来测试端点。我找不到提供任意 URL 参数的方法。我很高兴将它们作为单个字符串输入。

人们可能会这样想：

def books(**params): 
    ....

这给出了一个curl语句：

api/books?params=sort%5Bone%5D%3Dtwo'

我真正想要的是：

api/books?sort&one%5D%3Dtwo'

我正在使用 FastAPI 和 Structlog，想要测试日志格式并将其从纯文本/字符串转换为 JSON 格式，以便日志聚合器平台具有更好的可读性和处理能力。面临某些日志输出以 JSON 形式提供但以纯字符串形式存在的情况。

电流输出

INFO:     127.0.0.1:62154 - "GET /api/preface HTTP/1.1" 200 OK
INFO:     127.0.0.1:62154 - "GET /loader.json HTTP/1.1" 200 OK
INFO:     127.0.0.1:62155 - "GET /hello_world HTTP/1.1" 200 OK
{"key":"test_key","message":"Push to NFS Success","event":"Testing Fast API..","logger":"test_my_api","filename":"main.py","func_name":"Hello_World","process":23760,"module":"docker","thread":23140,"pathname":"D:\\my_work\\fast_api\\main.py","process_name":"SpawnProcess-1","level":"info","time-iso":"2023-06-30T15:25:03.113400Z"}

预期输出：

    {
    "level": "INFO",
    "IP": "127.0 .0 .1: 62154",
    "method": "GET",
    "endpoint": "/loader.json",
    "protocol": "HTTP / 1.1",
    "status_code": 200,
    "status": "OK"
}
    {
    "level": "INFO",
    "IP": "127.0 .0 .1: 62155",
    "method": "GET",
    "endpoint": "/api/preface",
    "protocol": "HTTP / 1.1",
    "status_code": 200,
    "status": …

当 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执行语句后运行。

有什么解决方法吗？

我想在 FastAPI 中创建一个可能接收（多部分）Form数据或JSON正文的端点。有没有办法让这样的端点接受或者检测正在接收哪种类型的数据？

我有一个 api.py 文件，其中包含以下内容：

from fastapi import FastAPI
import logging
import uvicorn

app = FastAPI(title="api")

LOG = logging.getLogger(__name__)
LOG.info("API is starting up")
LOG.info(uvicorn.Config.asgi_version)

@app.get("/")
async def get_index():
    LOG.info("GET /"
    return {"Hello": "Api"}

该应用程序在本地运行：

uvicorn api:app --reload

INFO:     Will watch for changes in these directories: ['/Users/user/code/backend/api']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [44258] using StatReload
INFO:     Started server process [44260]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

它不记录任何启动消息。

稍后向 api 发送 http 请求时： …

我正在尝试将后台任务添加到我的中间件中，但尚未在他们的文档中找到任何官方方法来执行此操作，这是我迄今为止尝试过的方法：

async def task1():
    logging.info("Waiting ...")
    time.sleep(5)
    logging.info("Waited.")

@app.middleware("http")
async def add_process_time_header(request, call_next, bt=Depends(BackgroundTasks)):
    a = bt.dependency()
    a.add_task(task1)
    a()

    return await call_next(request)

这阻止了我的请求，我是否应该能够在没有等待的情况下调用异步函数？我想完全忽略后台任务的结果，我不需要 bg 任务的结果

此代码适用于 Flask

import requests
from flask import Flask,request,session,render_template, jsonify
import requests,time
from datetime import datetime,timedelta

app = Flask(__name__,template_folder='./')

@app.route('/')
def home():
    return render_template("For_putting_lat_lon.html")

@app.route('/distance') 
def check():
    Latitude1 = request.args.get("Latitude1")
    Latitude2 = request.args.get("Latitude2")  
    Longitude1 = request.args.get("Longitude1")
    Longitude2 = request.args.get("Longitude2")
    
    data = list()
    result = Travel_distance(Latitude1, Longitude1, Latitude2, Longitude2,data)
    
    if(result == 0):
        return render_template("noavailable.html")
    return render_template("For_posting_lat_lon.html",data = data)


def Travel_distance(Latitude1, Longitude1, Latitude2, Longitude2, data):
  geolocator = Nominatim(user_agent="geoapiExercises")
  location_name1 = geolocator.geocode(str(Latitude1)+", " +str(Longitude1))                                         
  location_name2 = geolocator.geocode(str(Latitude2)+", "+str(Longitude2))  
  req = requests.get(f"""http://router.project-osrm.org/route/v1/drive/{Longitude1},{Latitude1};{Longitude2},{Latitude2}?overview=false""") #car,bike,foot
  route_1 = …