相关疑难解决方法(0)

我有以下代码：

import time
from fastapi import FastAPI, Request
    
app = FastAPI()
    
@app.get("/ping")
async def ping(request: Request):
        print("Hello")
        time.sleep(5)
        print("bye")
        return {"ping": "pong!"}

如果我在本地主机上运行我的代码 - 例如http://localhost:8501/ping- 在同一浏览器窗口的不同选项卡中，我得到：

Hello
bye
Hello
bye

代替：

Hello
Hello
bye
bye

我已经阅读过有关使用的内容httpx，但仍然无法实现真正​​的并行化。有什么问题？

我有一个 FastAPI 应用程序，在多种不同的场合，需要调用外部 API。我使用 httpx.AsyncClient 进行这些调用。关键是我不完全理解应该如何使用它。

从httpx 的文档中我应该使用上下文管理器，

async def foo():
    """"
    I need to call foo quite often from different 
    parts of my application
    """
    async with httpx.AsyncClient() as aclient:
        # make some http requests, e.g.,
        await aclient.get("http://example.it")

但是，我明白，通过这种方式，每次我调用 时都会生成一个新客户端foo()，而这正是我们首先希望通过使用客户端来避免的情况。

我想另一种选择是在某处定义一些全局客户端，然后在需要时导入它，就像这样

aclient = httpx.AsyncClient()

async def bar():
    # make some http requests using the global aclient, e.g.,
    await aclient.get("http://example.it")

不过，第二个选项看起来有点可疑，因为没有人负责关闭会话等。

所以问题是：如何httpx.AsyncClient()在 FastAPI 应用程序中正确（重新）使用？

我有一个相对简单的 FastAPI 应用程序，它接受查询并从 ChatGPT 的 API 流回响应。ChatGPT 正在流回结果，我可以看到它在输入时被打印到控制台。

不工作的是StreamingResponse通过 FastAPI 返回。相反，响应会一起发送。我真的不知道为什么这不起作用。

这是 FastAPI 应用程序代码：

import os
import time

import openai

import fastapi
from fastapi import Depends, HTTPException, status, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.responses import StreamingResponse

auth_scheme = HTTPBearer()
app = fastapi.FastAPI()

openai.api_key = os.environ["OPENAI_API_KEY"]


def ask_statesman(query: str):
    #prompt = router(query)
    
    completion_reason = None
    response = ""
    while not completion_reason or completion_reason == "length":
        openai_stream = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[{"role": "user", "content": query}],
            temperature=0.0,
            stream=True,
        )
        for …

我正在上一堂发送通知的课。初始化时，它涉及到与通知服务器的连接，这非常耗时。我在 FastAPI 中使用后台任务来发送通知，因为我不想因通知而延迟响应。下面是示例代码。

file1.py:
noticlient = NotificationClient()

@app.post("/{data}")
def send_msg(somemsg: str, background_tasks: BackgroundTasks):
    result = add_some_tasks(data, background_tasks, noticlient)
    return result

file2.py:
def add_some_tasks(data, background_tasks: BackgroundTasks, noticlient):
    background_tasks.add_task(noticlient.send, param1, param2)
    result = some_operation
    return result

这里，通知客户端是全局声明的。我可以在file2.pyunder中初始化它add_some_tasks，但每次请求到达时它都会被初始化，这需要一些时间。有什么办法可以使用中间件在每次请求到达时重新使用它，这样就不需要每次都进行初始化。

或方法二：在类 def 中初始化通知

file1.py:
class childFastApi(FastAPI):
    noticlient = NotificationClient()

app = childFastApi()

@app.post("/{data}")
def send_msg(somemsg: str, background_tasks: BackgroundTasks):
    result = add_some_tasks(data, background_tasks, app.noticlient)
    return result

例子

这是我的代码：

from typing import  List
from fastapi import FastAPI, File, UploadFile
import asyncio
import concurrent.futures

app = FastAPI()
@app.post("/send_images")
async def update_item(
    files: List[UploadFile] = File(...),
):
    return {"res": len(files)}

我向该服务器发送请求（这些特定网址仅作为示例）：

import requests
import os 
import json
import numpy as np
import time
import io
import httpx
import asyncio
from datetime import datetime
import pandas as pd

from random import shuffle
import pandas as pd
import urllib
import io
from PIL import Image
from matplotlib import pyplot as plt …

我能够从另一个 API 获得一个 API 的响应，但无法将其存储在某处（在返回响应之前的文件或其他内容中） response=RedirectResponse(url="/apiname/")（我想访问带有标头和正文的发布请求）

我想存储这个响应内容而不返回它。

是的，如果我返回函数，我会得到结果，但是当我打印它时，我找不到结果。另外，如果我给出发布请求，那么我会收到错误实体未找到。

我阅读了 starlette 和 fastapi 文档，但找不到解决方法。回调也没有帮助。

我有一个使用Uvicornwith的应用程序FastAPI。我还打开了一些连接（例如到MongoDB）。SIGINT一旦出现某些信号（ 、SIGTERM和） ，我想优雅地关闭这些连接SIGKILL。

我的server.py文件：

import uvicorn
import fastapi
import signal
import asyncio

from source.gql import gql


app = fastapi.FastAPI()

app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
app.mount("/graphql", gql)

# handle signals
HANDLED_SIGNALS = (
    signal.SIGINT,
    signal.SIGTERM
)

loop = asyncio.get_event_loop()
for sig in HANDLED_SIGNALS:
    loop.add_signal_handler(sig, _some_callback_func)

if __name__ == "__main__":
    uvicorn.run(app, port=6900)

不幸的是，我尝试实现这一目标的方法不起作用。当我尝试Ctrl+C在终端中时，没有任何反应。我相信这是因为Uvicorn是在不同的线程中启动的......

这样做的正确方法是什么？我注意到了uvicorn.Server.install_signal_handlers()这个功能，但使用起来并不幸运......

我有一个 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 请求时： …

我最近正在使用 fastapi，作为练习，我想将我的 fastapi API 与其他服务器上的验证服务连接...但我不知道如何做到这一点，我还没有在官方文档中找到对我有帮助的东西..我必须用python代码来做吗？或者有什么办法吗？

FastApi 文档

这个想法是从一个端点获取文件对象并将其发送到其他端点以使用它而不保存它。让我们看一下这个示例代码：

import httpx
from fastapi import Request, UploadFile, File


app = FastAPI()
client = httpx.AsyncClient()

@app.post("/endpoint/")
async def foo(request: Request, file: UploadFile = File(...))
    urls = ["/some/other/endpoint", "/another/endpoint/"]
    for url in urls:
        response = await client.post(url) # here I need to send the file to the other endpoint 
    return {"bar": "baz"}


@app.post("/some/other/endpoint/")
async def baz(request: Request, file: UploadFile = File(...)): # and here to use it
     # Do something with the file object
     return {"file": file.filename}


@app.post("/another/endpoint/")
async def baz(request: …

我正在尝试使用 python Fast API 框架代理外部网站（在不同容器上运行的 Flower 监控 URL）：

client = AsyncClient(base_url=f'http://containername:7800/monitor')\n\n@app.get(\xe2\x80\x9c/monitor/{path:path}\xe2\x80\x9d)\nasync def tile_request(path: str):\n    req = client.build_request("GET", path)\n    r = await client.send(req, stream=True)\n    return StreamingResponse(\n        r.aiter_raw(),\n        background=BackgroundTask(r.aclose),\n        headers=r.headers\n   )\n

它能够代理每个路径的容器 URL。对于前。

http://python_server:8001/monitor/dashboard --> http://containername:7800/monitor/dashboard

http://python_server:8001/monitor/tasks --> http://containername:7800/monitor/tasks

效果很好。但是当 PATH 的 URL 中有一些查询参数时，它会失败。

对于前。

http://python_server:8001/monitor/dashboard?json=1&_=1641485992460 --> redirects to http://containername:7800/monitor/dashboard \n

（请注意，URL 中不会附加任何查询参数）。

任何人都可以帮助我们如何使用任何查询参数代理此外部网站的任何路径。

我有一个用于测试/开发目的的 FastAPI 应用程序。我想要的是，到达我的应用程序的任何请求都会按原样自动发送到另一台服务器上的另一个应用程序，并具有完全相同的参数和相同的端点。这不是重定向，因为我仍然希望应用程序像往常一样处理请求并返回值。我只想向不同服务器上的不同版本的应用程序发起类似的请求，而不需要等待其他服务器的答复，以便其他应用程序获取该请求，就像原始请求发送给它一样。

我怎样才能做到这一点？以下是我用于处理请求的示例代码：

@app.post("/my_endpoint/some_parameters")
def process_request(
    params: MyParamsClass,
    pwd: str = Depends(authenticate),
):
    # send the same request to http://my_other_url/my_endpoint/
    return_value = process_the_request(params)
    return return_value.as_json()

我正在使用httpx库，但我认为aiohttp的原理是相同的。如果我在应用程序的整个生命周期中为多个请求创建并重用 AsyncClient，我是否需要在应用程序关闭事件时调用aclose()（或者如果使用 Client）？close或者这些联系会自行消失。

如果我在 Docker 容器中运行应用程序会怎样？这也会是一个因素吗？

我不明白 AsyncClient 或 Client （或 aoihttp 中的 ClientSession）对象下面发生了什么。

感谢帮助。