Сохранение fastAPI POST в csv

У меня есть метеостанция, которая отправляет данные в формате json, и я хочу, чтобы сервер fastAPI принимал их и сохранял на диск. В настоящее время у меня есть

from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class WeatherItem(BaseModel):
    wind_direction_raw: int
    rain_amount_raw: int
    timestamp: list = []
    elapsed_time: int
    wind_speed_raw: int
    message_id: int

@app.post("/station")
async def create_item(item: WeatherItem):
    return item

Что хорошо работает, когда я запускаю его с uvicorn main:app --host 192.168.1.151 --port 9494 и отправляю тестовые данные из curl с помощью

curl -X POST -H "Content-Type: application/json" --data '{"elapsed_time": 6245, "timestamp": [2020, 7, 26, 12, 2, 21, 6, 208], "wind_direction_raw": 108, "wind_speed_raw": 5, "message_id": 666, "rain_amount_raw": "0"}' http://192.168.1.151:9494/station

Теперь мне нужно сохранить эти данные на диск. Я думаю, что самым простым было бы добавить его в файл csv. Но я не могу понять, как экспортировать пидантическую модель в csv. Есть ли простой способ сделать это или предпочтительны другие методы сериализации? Я хотел бы проанализировать эти данные в R, поэтому они мне нужны на диске в достаточно взаимозаменяемом формате.


python fastapi pydantic
person pgcudahy    schedule 11.02.2021    source источник

Ответы (2)


arrow_upward
1
arrow_downward

Насколько мне известно, у вас есть следующие варианты:

Опция 1

Добавьте прямые значения в файл CSV, перечислив весь список таких записей

with open("myFile.csv", "a") as f:
    f.write(f"{model.entry},{model.another_entry}")

(это версия с блокировкой)

Следует отметить, что вы также можете создать словарь и перебирать значения, добавлять их, хотя трудно сохранить порядок между записями.

Вариант 2

Загрузите файл в память с помощью pandas, добавьте в DataFrame, а затем сохраните данные в файл

import pandas as pd

...

data = pd.read_csv("my_file.csv")
data_to_add = DataFrame.from_dict(my_model.dict())
data = data.append(data_to_add)
data.to_csv("my_file.csv", index=False)
person lsabi    schedule 11.02.2021

arrow_upward
0
arrow_downward

У @Isabi есть хороший ответ, который направил меня на верный путь, и я его принял. Но для полноты картины я пошел с

import csv

...

@app.post("/station")
async def create_item(item: WeatherItem):

    write_path = "/home/pi/weather_station/data/weather_data_" + date.today().strftime("%Y-%m-%d") + ".csv"

    with open(write_path, "a") as f:
        writer = csv.writer(f)
        writer.writerow(dict(item).values())
    return item
person pgcudahy    schedule 12.02.2021