きり丸の技術日記

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

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

始めに

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のファイルアップロード時にBaggageSentry-Traceのリクエストヘッダが付与されてしまったため。また、S3側でAccess-Control-Allow-Headersを指定していたので、BaggageSentry-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で検証しないといけないということを痛感しました。

参考情報

PydanticのJsonはそのまま使わないほうがいい

始めに

PydanticにはJsonという便利な型があります。便利ではあるのですが、素のJson型では開発しづらい点があったので自分のアプリケーションでは拡張して使っています。

今回の記事では何に困ったのか、どのように拡張したのかを記載します。

環境

  • Python
    • 3.12.4
  • FastAPI
    • 0.114.2
  • Pydantic
    • 2.9.1

ユースケース

  • フロントで自由なデータ構造を定義してバックエンドとしては単純に保存する。また、フロントで取り出せるようにバックエンドで保存したデータ構造をフロントに返却する。

困ったケース

PydanticJsonで定義するとフロントで定義した自由なデータ構造がJson型であることを保証してくれます。ですので、フロントからバックエンドへの接続時には問題ありません。問題は、バックエンドからフロントにデータ構造を返却した時でした。PythonとしてはJsonとはListDictにあたります。

# 次はすべてJson
{}
[1, 2, 3]
{'value': {'nestedKey': 'value'}}

そのため、フロントから得たデータ型をそのまま返却した時に次のエラーが出力されてしまいます。

pydantic_core._pydantic_core.ValidationError: 2 validation errors for _Test3
value.json[any]
  JSON input should be string, bytes or bytearray [type=json_type, input_value={'value': [{'a': 'b'}]}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.8/v/json_type
value.is-instance[JSON]
  Input should be an instance of JSON [type=is_instance_of, input_value={'value': [{'a': 'b'}]}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.8/v/is_instance_of

一応、毎回json.dumpsで文字列に変換すれば問題はないのですが、影響範囲がかなり大きい状態になっていたので型で解決することにします。

実装

PydanticJsonだけでなく、DictListも定義しました。このように定義することで、フロントからバックエンドはデータを受け取りつつ、バックエンドからフロントに容易に返却できます。

JsonField = Annotated[
    Json[Any] | Dict[str, Any] | List[Any],
    Field(examples=['{"value": {"a": "b"}}', ["A", "B"], {'value': [{'a': 'b'}]}]),
]

一応、Jsonを残しておくメリットもないのですが、エラーメッセージはJsonのものを使用したいので、そのままにしています。

pydantic_core._pydantic_core.ValidationError: 3 validation errors for _Test2
value.json[any]
  Invalid JSON: key must be a string at line 1 column 2 [type=json_invalid, input_value='{[1, 2, 3]}', input_type=str]
    For further information visit https://errors.pydantic.dev/2.8/v/json_invalid
value.dict[str,any]
  Input should be a valid dictionary [type=dict_type, input_value='{[1, 2, 3]}', input_type=str]
    For further information visit https://errors.pydantic.dev/2.8/v/dict_type
value.list[any]
  Input should be a valid list [type=list_type, input_value='{[1, 2, 3]}', input_type=str]
    For further information visit https://errors.pydantic.dev/2.8/v/list_type

実装およびテストコードについては次のとおりです。

import json
import pytest
from typing import Annotated, Dict, Any, List

from pydantic import BaseModel, Json, Field

JsonField = Annotated[
    Json[Any] | Dict[str, Any] | List[Any],
    Field(examples=['{"value": {"a": "b"}}', ["A", "B"], {'value': [{'a': 'b'}]}]),
]

class TestJson:
    class _Test2(BaseModel):
        value: JsonField

    @pytest.mark.parametrize(
        "value",
        [
            '{"value": {"a": "b"}}',
            ["A, B"],
            {'value': [{'a': 'b'}]}
        ],
    )
    async def test_02(self, value: Any):
        # Dict と Listを定義しているので、json.dumpsは不要
        instance = self._Test2(value=value)
        self._Test2.model_validate(instance.model_dump())

別のメリット

また、型定義をしつつFieldexamplesを伝えることによって、OpenAPIのサンプルレスポンスがわかりやすくなります。内部的にはJSONであればなんでも受け取れるのですが、サンプルがないと事前にシリアライズした状態でリクエストしてほしいように見えます。

OpenAPIのJson型でexampleを指定していない場合
OpenAPIのJson型でexampleを指定していない場合にstringを期待しているように見える
OpenAPIのexamplesを付与する
examplesで自由にパラメータを指定できることを伝えられる

各項目でexamplesを定義するのは面倒ですので、Annotatedで定義箇所を1つにまとめられるのはメリットです。

ソースコード

終わりに

Json型をそのまま使うことはないかもしれませんが、覚えておくと少し便利です。

参考情報