相关疑难解决方法(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，但仍然无法实现真正​​的并行化。有什么问题？

我有一个.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)，它起作用了，因为它返回一个字符串。

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

我在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 …

我创建了一个端点，如下所示：

@app.post("/report/upload")
def create_upload_files(files: UploadFile = File(...)):
        try:
            with open(files.filename,'wb+') as wf:
                wf.write(file.file.read())
                wf.close()
        except Exception as e:
            return {"error": e.__str__()}

它是用 uvicorn 启动的：

../venv/bin/uvicorn test_upload:app --host=0.0.0.0 --port=5000 --reload

我正在执行一些测试，使用 Python 请求上传大约100 MB的文件，大约需要 128 秒：

../venv/bin/uvicorn test_upload:app --host=0.0.0.0 --port=5000 --reload

我使用 Flask 通过 API 端点测试了相同的上传脚本，大约需要 0.5 秒：

f = open(sys.argv[1],"rb").read()
hex_convert = binascii.hexlify(f)
items = {"files": hex_convert.decode()}
start = time.time()
r = requests.post("http://192.168.0.90:5000/report/upload",files=items)
end = time.time() - start
print(end)

我做错了什么吗？

我有一个 FastAPI 端点，它接收文件，将其上传到 s3，然后对其进行处理。除了处理失败外，一切正常，并显示以下消息：

  File "/usr/local/lib/python3.9/site-packages/starlette/datastructures.py", line 441, in read
    return self.file.read(size)
  File "/usr/local/lib/python3.9/tempfile.py", line 735, in read
    return self._file.read(*args)
ValueError: I/O operation on closed file.

我的简化代码如下所示：

async def process(file: UploadFile):
    reader = csv.reader(iterdecode(file.file.read(), "utf-8"), dialect="excel")  # This fails!
    datarows = []
    for row in reader:
        datarows.append(row)
    return datarows

如何读取上传文件的内容？

更新

我设法进一步隔离问题。这是我的简化端点：

import boto3
from loguru import logger
from botocore.exceptions import ClientError


UPLOAD = True

@router.post("/")
async def upload(file: UploadFile = File(...)):
    if UPLOAD:
        # Upload the file
        s3_client = …

我已经使用 tweepy 将推文的文本存储在使用 Python 的 csv 文件中csv.writer(），但我必须在存储之前以 utf-8 对文本进行编码，否则 tweepy 会抛出一个奇怪的错误。

现在，文本数据存储如下：

"b'Lorem Ipsum\xc2\xa0Assignment '"

我尝试使用此代码对其进行解码（其他列中有更多数据，文本在第 3 列中）：

with open('data.csv','rt',encoding='utf-8') as f:
    reader = csv.reader(f,delimiter=',')
    for row in reader:
        print(row[3])

但是，它不会解码文本。我不能使用，.decode('utf-8')因为 csv 阅读器将数据作为字符串读取，即type(row[3])是'str'，我似乎无法将其转换为bytes，数据再次被编码！

如何解码文本数据？

编辑：这是来自 csv 文件的示例行：

"b'Lorem Ipsum\xc2\xa0Assignment '"

注意：如果解决方案是在编码过程中，请注意我无法再次下载整个数据。

我有这个非常简单的代码，它获取字典作为输入并返回字典：

app = FastAPI()

class User(BaseModel):
    user_name: dict


@app.post("/")
def main(user: User):
   ## some processing
   return user

我用以下 python 代码调用它：

import requests
import json 
import pandas as pd

df = pd.read_csv("myfile.csv")
data = {}
data["user_name"] = df.to_dict()
headers = {'content-type': 'application/json'}
url = 'http://localhost:8000/'
resp = requests.post(url,data=json.dumps(data), headers=headers )
resp

现在，我想做类似的事情，但我不想通过 python 请求发送数据，而是上传本地文件，将其发送到 API 并获取处理后的 .csv 文件。

现在我有以下代码来上传文件：

from typing import Optional
from fastapi import FastAPI
from fastapi import FastAPI, File, UploadFile
from pydantic import BaseModel
from typing import List …

我的应用程序在前端使用 React，在后端使用 FastAPI。

我正在尝试将 csv 文件上传到我的服务器。

提交表单时，这会被调用：

  const onSubmit = async (e) => {
    e.preventDefault();
    const formData = new FormData();
    formData.append("file", file);
    fetch("/api/textitems/upload", {
      method: "POST",
      body: formData,
    });
  };

数据由以下人员接收：

@app.post('/api/textitems/upload')
def upload_file(csv_file: UploadFile = File(...)):
    dataframe = pd.read_csv(csv_file.file)
    return dataframe.head()

我不断收到INFO: 127.0.0.1:0 - "POST /api/textitems/upload HTTP/1.1" 422 Unprocessable Entity错误。

我能够使用curl 成功执行post 请求，如下所示：

curl -X POST "http://localhost:8000/api/textitems/upload" -H "accept: application/json" -H "Content-Type: multipart/form-data" -F "csv_file=@exp_prod.csv;type=text/csv"

关于我在使用 Javascript 时出错的地方有什么建议吗？

我有一个 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 …

这是面试中经常出现的问题。

我知道如何使用 .csv 文件读取 csv 文件Pandas。

但是，我正在努力寻找一种无需使用外部库即可读取文件的方法。

Python 是否带有任何有助于读取 csv 文件的模块？

我想在 fastapi 中运行一个简单的后台任务，它在将其转储到数据库之前涉及一些计算。然而，计算会阻止它接收更多请求。

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()
db = Database()

async def task(data):
    otherdata = await db.fetch("some sql")
    newdata = somelongcomputation(data,otherdata) # this blocks other requests
    await db.execute("some sql",newdata)
   


@app.post("/profile")
async def profile(data: Data, background_tasks: BackgroundTasks):
    background_tasks.add_task(task, data)
    return {}

解决此问题的最佳方法是什么？

我正在尝试遍历 csv 文件。但是，接收到的文件很难阅读。我搜索了这个，我找不到明确的解决方案！

@app.get("/uploadsequence/")
async def upload_sequence_form():
    return HTMLResponse("""
            <!DOCTYPE html>
            <html>
                <head>
                    <title>sequence upload</title>
                </head>
                <body>
                    <h1>upload sequence .CSV file</h1>
                    <form method='post' action='/uploadsequence/' enctype='multipart/form-data'>
                        Upload a csv file: <input type='file' name='csv_file'>
                        <input type='submit' value='Upload'>
                    </form>
                </body>
            </html>
            """)

@app.post("/uploadsequence/")
async def upload_sequence(csv_file: UploadFile = File(...), db = Depends(get_db)):
        csv_file_encoded = TextIOWrapper(csv_file.file, encoding='utf-8')
        csv_reader = csv.DictReader(csv_file_encoded)
        for row in csv_reader:
            if row["Well Type"] in ["RIG MAINTENANCE","RIG MOVE","RIG COMMISSIONING","ABANDONMENT","LEARNINGCURVE"]:
                crud.insert_sequence_record(db=db, row=row,is_drilling=False)
            else:
                crud.insert_sequence_record(db=db, row=row,is_drilling=True)

它给了我这个错误： csv_file_encoded = TextIOWrapper(csv_file.file, encoding='utf-8') AttributeError: 'SpooledTemporaryFile' …

我想将文件上传到 FastAPI 后端并将其转换为 Pandas DataFrame。但是，我似乎不明白如何使用 FastAPI 的UploadFile对象来做到这一点。更具体地说，我应该将什么传递给该pd.read_csv()函数？

这是我的 FastAPI 端点：

@app.post("/upload")
async def upload_file(file: UploadFile):
    df = pd.read_csv("")
    print(df)
    return {"filename": file.filename}