RevCommでバックエンド開発をしている小門です。

最近、CSVファイルのアップロードを受け付けて処理するバックエンドAPIの機能開発を担いました。
CSVファイルのパース、バリデーションにPydanticが便利でしたので紹介したいと思います。

なお開発言語はPython、コードの動作バージョンは以下です。

Python 3.12
Pydantic: 2.6.0

PythonでCSVファイルの取り扱い

Pythonでは組み込みモジュールcsvを使うことで基本的なCSVファイルの読み取り・書き込みができます。

# persons.csv の例
"""
"name","age"
"alice",20
"bob",21
"""

import csv
with open("persons.csv", newline="") as csvfile:
  reader = csv.DictReader(csvfile)
  for row in reader:
    print(row)

# {"name": "alice", "age": "20"}
# {"name": "bob", "age": "21"}

また、取り扱うCSVファイルのカラム形式が決まっている場合はその情報をクラスに定義することで仕様が明確になり、開発時にエディターの支援を受けられるメリットがあります。

これにはPythonの組み込み機能で3つの方法が考えられます。

typing.NamedTuple
typing.TypedDict
dataclass

from typing import NamedTuple, TypedDict
from dataclasses import dataclass

class Person(NamedTuple):
  name: str
  age: str

class Person(TypedDict):
  name: str
  age: str

@dataclass
class Person:
  name: str
  age: str


data = {"name": "Alice", "age": "20"}
# いずれのPersonクラスでも以下で初期化可能
person = Person(**data)

本機能のユースケース

今回開発したCSVファイルの機能では重要な考慮点が大きく2つありました。

データバリデーション

CSVファイルはサービスのエンドユーザーから直接アップロードされるものであるため、ファイルの中身を厳密にバリデーション（検証）する必要がありました。

csv.reader と csv.DictReader は文字列型でデータを読み取るため、上記の例においては age カラムも str 型で定義しました。
しかし実際の開発シーンではもちろん age は整数型で扱う必要があるでしょう。他にも浮動小数や日付などのデータを扱う場合はそれぞれ変換する必要があります。

またバリデーションとしてはデータ型以外にも値自体の検証や必須項目のチェックなども必要です。

動的なCSVヘッダー

MiiTelは海外にサービス展開しているため多言語対応しています（現在は日本語と英語）。
そのため、ユーザーの言語設定によってCSVファイルのヘッダー行（先頭行）が変化することが要件でした。

例

言語設定：日本語

"氏名","メールアドレス","管理者権限の有無"
"Kokado","shota.kokado@example.com","true"

言語設定：英語

"Name","Email","Administrative Privilege"
"Kokado","shota.kokado@example.com","true"

Pydanticの導入

以上の要件を実現するためには上述した組み込みモジュールの機能だけでは不十分と考え、ライブラリの利用を検討しました。結論としてはあまり悩まずにPydanticを採用することに決めました。

ポイント:

型アノテーションを活用して型変換やリッチなバリデーションを少ないコードで実装できる
FastAPIで採用されており利用事例が多く、ライブラリの信頼度が高い

データバリデーション

Pydanticは公式ドキュメントで謳われている通り型・データのバリデーションが主機能の一つであるため、ダイレクトに恩恵を享受することができました。

Pydanticを使うと冒頭の例は以下のように実装できます。

from pydantic import BaseModel

class Person(BaseModel):
  name: str
  age: int  # <== str ではなく int

data = {"name": "Alice", "age": "20"}
person = Person(**data)
print(person.name, type(person.name))
# Alice <class 'str'>
print(person.age, type(person.age))
# 20 <class 'int'>

また発展として、Enum型も活用することで入力値に意味を持たせることができるようになります。

from enum import IntEnum

class Sex(IntEnum):
  MALE = 0
  FEMALE = 1

class Person(BaseModel):
  name: str
  age: int
  sex: Sex  # <==

data = {"name": "Alice", "age": "20", "sex": "1"}
person = Person(**data)
print(person.sex)
# <Sex.FEMALE: 1>

動的なCSVヘッダー

動的なCSVヘッダーを受け付けるためのPydanticの使い方として2つの方法を考えました。

CSV例を再掲します。

言語設定：日本語

"氏名","メールアドレス","管理者権限の有無"
"Kokado","shota.kokado@example.com","true"

言語設定：英語

"Name","Email","Administrative Privilege"
"Kokado","shota.kokado@example.com","true"

一つ目は各言語ごとのCSVファイルを表現するためのPydanticモデルクラスをそれぞれ定義することです。

from pydantic import BaseModel, Field

# 1. 言語設定：日本語用
class CsvDataJa(BaseModel):
  name: str = Field(alias="氏名")
  email: str = Field(alias="メールアドレス")
  is_administrator: bool = Field(alias="管理者権限の有無")

# 1. 言語設定：英語用
class CsvDataEn(BaseModel):
  name: str = Field(alias="Name")
  email: str = Field(alias="Email")
  is_administrator: bool = Field(alias="Administrative Privilege")

2つ目はpydantic.AliasChoicesを使って複数のエイリアスを許容するようにして一つのPydanticモデルクラスを定義することです。

Alias - Pydantic

from enum import Enum
from pydantic import AliasChoices, BaseModel, Field

class CsvField(Enum):
  # (ja, en)
  NAME = ("氏名", "Name")
  EMAIL = ("メールアドレス", "Email")
  ADMINISTRATIVE_PRIVILEGE = ("管理者権限の有無", "Administrative Privilege")

class CsvData(BaseModel):
  name: str = Field(validation_alias=AliasChoices(*CsvField.NAME.value))
  email: str = Field(validation_alias=AliasChoices(*CsvField.EMAIL.value))
  is_administrator: bool = Field(validation_alias=AliasChoices(*CsvField.ADMINISTRATIVE_PRIVILEGE.value))

結論としては2つ目の方法を採用しました。クラス毎の責務を適切に分割できたと考えたためです。

CsvField: 言語毎のCSVファイルのカラム名を管理する
CsvData: CSVカラム毎のデータ型、バリデーションロジックを管理する

例えば新しい別の言語に対応する必要が出た場合は CsvField クラスのみ、既存カラムのバリデーションロジックの仕様変更する際は CsvData クラスのみの更新で済みます。

ただ一点、AliasChoicesは利用可能な複数のaliasを定義するのみであるため、Field同士でaliasの組み合わせは任意になることが注意事項です。
今回の例では日本語、英語の組み合わせを許容することになります。

from .types.csv_data import CsvData  # 上述のCsvDataクラス

# 以下のdataから等価なCsvDataが作られる
data = {"氏名": "Alice", "メールアドレス": "xxx@example.com", "管理者権限の有無": "true"}
data = {"氏名": "Alice", "Email": "xxx@example.com", "Administrative Privilege": "true"}

csv_data = CsvData(**data)
print(csv_data)
# CsvData(name='Alice', email='xxx@example.com', is_administrator=True)

以上を踏まえて、CSVの読み込みを行う処理を以下のように実装しました。
※同時にバリデーションロジックも実際をイメージしたものにアップデート

# types/csv_data.py
from enum import Enum
from pydantic import AliasChoices, BaseModel, EmailStr, Field

class CsvField(Enum):
  NAME = ("氏名", "Name")
  EMAIL = ("メールアドレス", "Email")
  ADMINISTRATIVE_PRIVILEGE = ("管理者権限の有無", "Administrative Privilege")

class CsvData(BaseModel):
  name: str = Field(validation_alias=AliasChoices(*CsvField.NAME.value, min_length=1))
  email: EmailStr = Field(validation_alias=AliasChoices(*CsvField.EMAIL.value))
  is_administrator: bool = Field(validation_alias=AliasChoices(*CsvField.ADMINISTRATIVE_PRIVILEGE.value))

# main.py
import csv
from pydantic_core import ValidationError

from .types.csv_data import CsvData

filename = "path/to/file.csv"
with open(filename, newline="") as csvfile:
  reader = csv.DictReader(csvfile)
  for row in reader:
    try:
      csv_data = CsvData(**row)
    except ValidationError:
      handle_error()
      raise

    # 以降の処理

まとめ

CSVファイルを扱うバックエンド機能の開発でPydanticを活用した事例を紹介しました。

データの型変換やバリデーションに関する処理を自前で実装する量が減り、サービスのドメインロジックに基づく実装に集中することができました。
Pydanticは他にもJSON形式のシリアライズ、エクスポートなど豊富な機能を持っているので使いこなしていきたいです。

RevComm Tech Blog

コミュニケーションを再発明し人が人を想う社会を創る

Pydanticを活用してCSVファイルを型安全に扱う

PythonでCSVファイルの取り扱い

本機能のユースケース

データバリデーション

動的なCSVヘッダー

Pydanticの導入

データバリデーション

動的なCSVヘッダー

まとめ