Блог

У меня есть метеостанция, которая отправляет данные в формате 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. Но я не могу понять, как экспортировать модель pydantic в csv. Есть ли простой способ сделать это или предпочтительнее другие методы сериализации? Я хотел бы проанализировать эти данные, R поэтому они мне нужны на диске в довольно взаимозаменяемом формате.

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

Вариант 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)
 

у @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