使用 Pydantic 创建 CSV 行

Question

我们使用 Pydantic 来设置“域模型”并在应用程序层中使用它们。

import datetime
from decimal import Decimal

from pydantic import BaseModel


class Person(BaseModel):
    name: str
    birth_date: datetime.date
    height: Decimal
    # other 100 fields, most of them Decimal

现在我们需要将其中一个模型导出到 CSV，第一个实现很简单：

import csv
from typing import Iterable


def store(persons: Iterable[Person]):
    fieldnames = list(Person.schema()["properties"].keys())
    
    with open("/tmp/test.csv", "w") as fp:
        writer = csv.DictWriter(fp, fieldnames=fieldnames)
        writer.writeheader()
        for person in persons:
            writer.writerow(person.dict())

但还有更多，因为我们希望所有小数都限制为小数点后两位，我们怎样才能实现这一点呢？

注意：不会将序列化逻辑放入域模型中，但我愿意为序列化创建一个新模型。

Answer 1

您可以使用自定义 json 编码器对模型中的所有小数进行四舍五入（不幸的是，它不能用作Decimal结果类型，json_encoders因为它不是 JSON 可序列化的。

我还定义了一个PersonOut继承自的模型Person，因此您不必将序列化逻辑存储在域模型中。在行写入器中，模型映射到序列化模型。

import csv
import json
from decimal import Decimal
from pydantic import BaseModel
from typing import Iterable


class Person(BaseModel):
    """ Person domain model. """
    name: str
    height: Decimal


class PersonOut(Person):
    """ Person model used for serialization. """
    class Config:
        json_encoders = { Decimal: lambda v: float(round(v, 2)) }


def store(persons: Iterable[Person]):
    fieldnames = list(Person.schema()["properties"].keys())
    
    with open("test.csv", "w") as fp:
        writer = csv.DictWriter(fp, fieldnames=fieldnames)
        writer.writeheader()
        for person in persons:
            writer.writerow(json.loads(PersonOut(**person.dict()).json()))

if __name__=="__main__":
    persons = [Person(name='test', height=1.743)]

    store(persons)