小编Chr*_*ris的帖子

我通常使用 Tornado，并尝试迁移到 FastAPI。

假设我有一个非常基本的 API，如下所示：

@app.post("/add_data")
async def add_data(data):
    return data

当我运行以下 Curl 请求时： curl http://127.0.0.1:8000/add_data -d 'data=Hello'

我收到以下错误：

{"detail":[{"loc":["query","data"],"msg":"field required","type":"value_error.missing"}]}

所以我确信我错过了一些非常基本的东西，但我不知道那可能是什么。

我有一个.csv文件想要在 FastAPI 应用程序中呈现。我只设法.csv以 JSON 格式呈现文件，如下所示：

def transform_question_format(csv_file_name):

    json_file_name = f"{csv_file_name[:-4]}.json"

    # transforms the csv file into json file
    pd.read_csv(csv_file_name ,sep=",").to_json(json_file_name)

    with open(json_file_name, "r") as f:
        json_data = json.load(f)

    return json_data

@app.get("/questions")
def load_questions():

    question_json = transform_question_format(question_csv_filename)

    return question_json

当我尝试直接返回时pd.read_csv(csv_file_name ,sep=",").to_json(json_file_name)，它起作用了，因为它返回一个字符串。

我应该如何进行？我相信这不是一个好方法。

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

@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和Flask中编写了具有相同功能的相同 API 应用程序。但是，当返回 JSON 时，两个框架之间的数据格式不同。两者都使用相同的json库，甚至相同的代码：

import json
from google.cloud import bigquery
bigquery_client = bigquery.Client()

@router.get('/report')
async def report(request: Request):
    response = get_clicks_impression(bigquery_client, source_id)
    return response

def get_user(client, source_id):
    try:
        query = """ SELECT * FROM ....."""
        job_config = bigquery.QueryJobConfig(
            query_parameters=[
                bigquery.ScalarQueryParameter("source_id", "STRING", source_id),
            ]
        )
        query_job = client.query(query, job_config=job_config)  # Wait for the job to complete.
        result = []
        for row in query_job:
            result.append(dict(row))
        json_obj = json.dumps(result, indent=4, sort_keys=True, default=str)

    except Exception as e:
        return …

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实际发送到服务器的内容，因此请求不匹配。有任何想法吗？

即使正在发送有效的请求，我的POST请求仍然会失败并收到响应。我正在尝试创建一个网络应用程序，该应用程序接收上传的带有各种遗传标记的文本文件，并将其发送到张量流模型以进行癌症生存预测。可以在此处找到 github 项目的链接。422JSON

这是POST请求：

 df_json = dataframe.to_json(orient='records')
 prediction = requests.post('http://backend:8080/prediction/', json=json.loads(df_json), headers={"Content-Type": "application/json"})

这是 pydantic 模型以及 API 端点：

class Userdata(BaseModel):
RPPA_HSPA1A : float
RPPA_XIAP : float
RPPA_CASP7 : float
RPPA_ERBB3 :float
RPPA_SMAD1 : float
RPPA_SYK : float
RPPA_STAT5A : float
RPPA_CD20 : float
RPPA_AKT1_Akt :float
RPPA_BAD : float
RPPA_PARP1 : float
RPPA_MSH2 : float
RPPA_MSH6 : float
RPPA_ACACA : float
RPPA_COL6A1 : float
RPPA_PTCH1 : float
RPPA_AKT1 : float
RPPA_CDKN1B : float …

我有这个模型：

class Text(BaseModel):
    id: str
    text: str = None


class TextsRequest(BaseModel):
    data: list[Text]
    n_processes: Union[int, None]

所以我希望能够接受如下请求：

{"data": ["id": "1", "text": "The text 1"], "n_processes": 8} 

和

{"data": ["id": "1", "text": "The text 1"]}.

现在在第二种情况下我得到

{'data': [{'id': '1', 'text': 'The text 1'}], 'n_processes': None}

使用此代码：

app = FastAPI()

@app.post("/make_post/", response_model_exclude_none=True)
async def create_graph(request: TextsRequest):
    input_data = jsonable_encoder(request)

n_processes那么这里我该如何排除呢？

我正在使用 FastAPIRedirectResponse并尝试将用户从一个应用程序（域）重定向到另一个应用程序（域），并在 ; 中设置一些responsecookie 然而，cookie 总是被删除/不被传输。如果我尝试添加一些标头，我添加到的所有标头RedirectResponse也不会传输。

@router.post("/callback")
async def sso_callback(request: Request):
   jwt_token = generate_token(request)
   redirect_response = RedirectResponse(url="http://192.168.10.1/app/callback", 
                             status_code=303)
   redirect_response.set_cookie(key="accessToken", value=jwt_token, httponly=True)
   redirect_response.headers["Authorization"] = str(jwt_token)
   return redirect_response

我该如何解决这个问题？先谢谢您的帮助。

我正在尝试在 Fastapi-users pydantic 架构中构建自定义字段，如下所示：

class UserRead(schemas.BaseUser[uuid.UUID]):
    twitter_account: Optional['TwitterAccount']

在 UserRead 验证上 Pydantic 返回

field required (type=value_error.missing)

'TwitterAccount' schema.update_forward_refs()最后调用每个字段。

TwitterAccount本身具有必填字段，并使它们可选并不是可接受的解决方法。我注意到我可以做到Optional[List['TwitterAccount']]并且会起作用，但这有点愚蠢。