Cerberusでrow（list of dicts、テーブルの複数行データ）を一気に検証する方法

やりたいこと

infoテーブル

name	company	telno
satoh	A	12345678
suzuki	B	87654321

テーブルの行みたいな、複数件のデータを一気にCerberusに流して検証したい

プログラミング上はこんな感じなもの

{'info': [{'name': 'satoh', 'company': 'A', 'telno': '12345678'}, {'name': 'suzuki', 'company': 'B', 'telno': '87654321'}]}

前提

・Pythonインストール済み
・pipenvインストール済み
・今回は便宜上pipenv内にtoxでpytestをやるですが、ローカルでpytestやcerberusなど諸々ライブラリを入れてテストをやっても大丈夫です

ディレクトリ構成

root 　　   
  ┣━ functions
      ┗━ api
          ┗━ cerberus_validator.py
  ┣━ tests
      ┗━ test_cerberus.py
  ┣━ Pipfile
  ┗━ tox.ini

ソースの中身

cerberus_validator.py

cerberus_validator.py

from cerberus import Validator

# まずテーブルの列の検証スキーマを定義
SCHEMA_INFO = {
    'name': {
        'type': 'string',
        'required': True,
        'empty': False,
        'nullable': False
    },
    'company': {
        'type': 'string',
        'required': True,
        'empty': False,
        'nullable': False
    },
    'telno': {
        'type': 'string',
        'required': True,
        'empty': False,
        'nullable': False
    },
}

# ここが肝心！！
# infoという項目の中にlist（配列）が入ることを定義
# その配列の子項目がdict形式（テーブルの行相当）であることを定義
# 最後はさきほど定義したテーブルの列の検証スキーマを教えてあげる
SCHEMA_INFO_LIST = {
    'info': {
        'type': 'list',
        'required': True,
        'schema': {
            'type': 'dict',
            'schema': SCHEMA_INFO,
        },
    }
}

def validate_info(info):
    # ここでは複数行のデータが入ったinfoを一気流す
    validator = Validator(allow_unknown=True)
    if not validator.validate(info, SCHEMA_INFO_LIST):
        # 失敗時はエラー文を出す
        print(validator.errors)
        return False
    return True

test_cerberus.py

test_cerberus.py

import importlib
import pytest


class Test:

    @pytest.mark.it("正常系")
    def test_1(self):
        from functions.api import cerberus_validator
        importlib.reload(cerberus_validator)
        
        info = {'info': [{'name': 'satoh', 'company': 'A', 'telno': '12345678'}, {'name': 'suzuki', 'company': 'B', 'telno': '87654321'}]}

        result = cerberus_validator.validate_info(info)

        assert result is True

    @pytest.mark.it("異常系")
    def test_2(self):
        from functions.api import cerberus_validator
        importlib.reload(cerberus_validator)

        # 欠損しているデータを入れる
        info = {'info': [{'name': '', 'company': 'A', 'telno': '12345678'}, {'name': 'suzuki', 'company': '', 'telno': '87654321'}]}

        result = cerberus_validator.validate_info(info)

        assert result is False

テスト実行

ただのpythonコマンドやpytestコマンドで動かしても大丈夫

pipenv run tox

すると、ちゃんと欠損してるデータを検知してくれた！
エラー箇所やエラー理由もちゃんと出したね！

終わりに

私は本来この書き方を知らずに、地味でforループ各行をチェックしてしまった。。。
もっと効率的な方法があることを知れてよかった！