きり丸の技術日記

技術検証したり、資格等をここに残していきます。

PydanticでEmailStrを拡張する

Python 技術

始めに

pydanticにはEmailStrというemailを検証するための拡張クラスがあります。しかし、Emailの仕様としてはUTF-8を許容しているものの、システム的にはASCIIしか許容したくないことがあります。その場合に向けて、EmailStrを継承してASCIIのみ許容する拡張クラスを作ります。

環境

  • Python
    • 3.12.7
  • Pydantic
    • 2.9.2

実装

まずは、ASCII文字だけチェックできるように正規表現を用意します。

^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$

次のコードでEmailStrを継承しつつ、自分が追加したvalidationを実行させます。

from typing import Type, Any
import re

from pydantic import EmailStr, validator, GetCoreSchemaHandler
from pydantic_core import core_schema


class CustomEmailStr(EmailStr):

    @classmethod
    def validate_half_and_full_email(cls, value: str) -> str:
        if not re.search(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$", value):
            raise ValueError("ERROR.EMAIL_VALIDATION")
        return value

    @classmethod
    def __get_pydantic_core_schema__(cls, source_type: Type[Any], handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
        email_schema = handler.generate_schema(EmailStr)
        return core_schema.chain_schema(
            [
                email_schema,
                core_schema.no_info_after_validator_function(
                    cls.validate_half_and_full_email,
                    core_schema.str_schema()
                ),
            ]
        )

このように定義しておくことで、各クラスで別々に追加バリデーションを定義する必要がなくなります。もちろん、処理を追加している分処理が遅くはなっているので、データが登録されうる最小限の箇所でのみ使用し、それ以外はデフォルトのEmailStrをそのまま使用しても問題ありません。

class _Test(BaseModel):
    email: CustomEmailStr

class TestIsValid:
    @pytest.mark.parametrize(
        "value, expected",
        [
            ("aiueo@example.com", True),
            ("AIUEO@example.com", True),
            ("aiueo+０@example.com", False),
            ("aiueo+あいうえお@example.com", False),
            ("あいうえお+あいうえお@example.com", False),
        ],
    )
    def test_is_valid(self, value, expected):
        if expected:
            _Test(email=value)
        else:
            with pytest.raises(ValidationError):
                _Test(email=value)

ソースコード

終わりに

単純にValidationをかけるパターンはたくさん出てきますが、特定のクラスを継承するパターンして横展開の工数を減らしたい目的の記事は見つかりませんでした。ぜひ活用してください。

参考情報

Pythonでgroup_byしたいならdefaultdictを使う

Python 技術

始めに

Pythonでデータをグループ化する際、defaultdictを使用すると簡単かつ効率的に実装できます。この記事では、defaultdictを使ったgroup_byの実装方法と、itertools.groupbyとの違いについて解説します。

環境

  • Python 3.12.6

実装

defaultdictを使用すればシンプルに実装できます。

from collections import defaultdict

class TestGroupBy:
    class _Test:
        def __init__(self, user_id, group_id):
            self.user_id = user_id
            self.group_id = group_id

    @pytest.fixture
    def parameters(self):
        return [
            self._Test(1, 'A'),
            self._Test(2, 'A'),
            self._Test(3, 'B'),
            self._Test(4, 'A'),
            self._Test(5, 'B'),
        ]

    class TestDefaultDict:
        def test_group_by(self, parameters):
            # NOTE: defaultdictは dictと違い、Keyが存在しない場合にもKeyErrorを発生させません
            grouped_data = defaultdict(list)
            for user in parameters:
                grouped_data[user.group_id].append(user.user_id)
            expected = {'A': [1, 2, 4], 'B': [3, 5]}
            assert dict(grouped_data) == expected

itertools.groupbyは次のコードで実装できます。

    class TestItertools:
        def test_group_by(self, parameters):
            # NOTE: ソートがかかっていないと正しくgroup_byされない
            non_continuous_data = {k: [user.user_id for user in v] for k, v in groupby(parameters, key=attrgetter('group_id'))}
            expected = {'A': [4], 'B': [5]}
            assert non_continuous_data == expected

            sorted_users = sorted(parameters, key=attrgetter('group_id'))

            grouped_data = {k: [user.user_id for user in v] for k, v in groupby(sorted_users, key=attrgetter('group_id'))}
            expected = {'A': [1, 2, 4], 'B': [3, 5]}
            assert grouped_data == expected

差分

基本的にはdefaultdictで問題ありません。

itertools.groupbyの場合はコード内にも記載していますが、非連続なデータの場合は期待どおりにgroup_byされないパターンがあるので特別に採用したいユースケースはないです。大規模データを変換したいことはあるでしょうが、そのときはpandasとかpolars使っているでしょうし…。一応、メモリ効率に軍配があがるので、OOMが発生したらitertools.groupbyを使用することを考えてもよいと思います。

ソースコード

終わりに

groupbyという名前がついているのでitertools.groupbyを使用していたのですが、非連続なデータでは使用できないという点でハマってしまいました。

Pythonで自分のブログに来る人はいないかもしれませんが、ぜひハマらないように注意してください。

【障害】SentryのtracePropagationTargetsの指定を誤った

ポストモーテム 技術

Sentryの設定を軽い気持ちで変更したところ大規模障害につながってしまったので、もう忘れないようにするためのメモ。

環境

  • @sentry/angular-ivy
    • 7.144.0

発生事象

S3へのファイルアップロードに失敗した。

原因

tracePropagationTargetsの指定を誤ってしまい、S3のファイルアップロード時にBaggageとSentry-Traceのリクエストヘッダが付与されてしまったため。また、S3側でAccess-Control-Allow-Headersを指定していたので、BaggageとSentry-Traceが付与されているとCORSエラーが発生して、HttpStatus403になっていました。

詳細

複数の検証環境を作り、検証環境名をサードレベルドメインに指定していました。

そのため、今後のことを考慮して、サードレベルドメインを正規表現で適用していました。

{
  "tracePropagationTargets": [
    "localhost",
    "https://example.com",
    "https://.*.example.com",
  ]
}

しかし、S3にファイルアップロードするURLには判別しやすいように、検証環境名のドメインを入れていました。

# URLイメージ
https://s3.ap-southeast-2/20240516-temporary.example.com
https://s3.ap-southeast-2/20240516-temporary.dev.example.com

これがhttps://.*.example.comの正規表現に引っかかってしまい、ファイルアップロード時に不要なリクエストヘッダが付与されてしまいました。

なお、正規表現の対象をOriginだけとしたい、とかは無理そうです。元々、tracingOriginsというオプションがありましたが、今回指定したtracePropagationTargetsに変更されています。

The tracingOrigins option was renamed tracePropagationTargets and deprecated in version 7.19.0 of the JavaScript SDK. tracingOrigins will be removed in version 8.

対策

比較したいのはURL全体ではなく、Origin部分だけではあるので、1単語だけヒットするような正規表現に変更しました。

https://\w+.example.com

ソースコード

なし。

終わりに

S3へのファイルアップロードをテストしなかったので、引っかかってしまいました。アプリケーションの範囲では動作確認していましたが、外部ストレージの範囲まではテストしていなかったのが落ち度です。

命名規則が頭に入っていれば検知できたと思いますが、ちゃんとE2Eで検証しないといけないということを痛感しました。

参考情報