使用 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 编码器来舍入模型中的所有小数（不幸的是，在 json_encoders 中使用 Decimal 作为结果类型是行不通的，因为它不是JSON 可序列化。

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

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)

使用 Pydantic 创建 CSV 行

use Pydantic to create CSV line

python

csv

serialization

decimal

pydantic