使用 Pydantic 创建 CSV 行答案

【问题标题】：use Pydantic to create CSV line使用 Pydantic 创建 CSV 行
【发布时间】：2021-09-07 19:40:35
【问题描述】：

我们使用 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())

但还有更多，因为我们希望所有小数都限制在小数点后两位，我们如何才能做到这一点？

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

【问题讨论】：

标签： python csv serialization decimal pydantic

【解决方案1】：

您可以使用自定义 json 编码器对模型中的所有小数进行四舍五入（不幸的是，在 json_encoders 中使用 Decimal 作为结果类型是行不通的，因为它不是 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)

【讨论】：