一覧PG-20-1 · データ処理の組立て

入力の正規化と検証

注文の入金・出荷イベントを処理します。注文IDは全角・半角と英字の大小を同一視する仕様とし、受注済み→入金済み→出荷済みの順だけを許可します。最後に途中まで反映する逐次処理と、全件成功時にまとめて反映する処理を比較します。

説明

外部から受け取った文字列は、そのまま状態更新へ使わず、構造への変換、値の検証、表記の統一を行います。JSONとして読めることと、業務上使える値であることは別です。

注文の入金・出荷イベントを処理します。注文IDは全角・半角と英字の大小を同一視する仕様とし、受注済み→入金済み→出荷済みの順だけを許可します。最後に途中まで反映する逐次処理と、全件成功時にまとめて反映する処理を比較します。

1JSONを構造へ変換する

全角の注文IDと、周辺に空白のあるイベント名を含むJSONを読みます。

import json, unicodedata; raw = '{"id":" Ａ-01 ","event":" PAY ","amount":120}'; parsed = json.loads(raw)JSONを辞書へ変換; print(parsed)変換後の型と値

JSONの構文は正しいので辞書になりますが、IDとイベント名の表記はまだ揃っていません。金額120は数値として読み込まれています。

2項目ごとの規則で正規化

IDにNFKC、周辺空白の除去、大小文字の統一を適用します。

normalized = unicodedata.normalize("NFKC", parsed["id"])全角などの互換表記を正規化.strip().casefold()周辺空白と大小文字を統一; print(repr(parsed["id"]), normalized, normalized == "a-01"同じIDとして照合)

全角Ａを含む表記がa-01になります。これはIDの比較規則として選んだ変換で、どんな文章や識別子にも無条件で適用するものではありません。異なる元の表記が同じ値へ集まるため、重複の扱いも統一後の値を基準にします。

3値の検査を通したイベントを返す

JSONを辞書へ変換した後で、必要な項目の型と範囲を検査します。成功時だけ正規化済みの辞書を返します。

def parse_event(text): item = json.loads(text)構造へ変換 if not isinstance(item, dict): raise ValueError("object required") if not isinstance(item.get("id"), str) or not isinstance(item.get("event"), str): raise ValueError("id and event must be strings") if type(item.get("amount")) is not int整数型を要求 or item["amount"] < 0: raise ValueError("amount must be a nonnegative integer") identity = unicodedata.normalizeIDの表記を統一("NFKC", item["id"]).strip().casefold() event = item["event"].strip().casefold()イベント名を統一 if not identity or event not in ("pay", "ship"): raise ValueError("invalid id or event") return {"id": identity検証済みの値を返す, "event": event, "amount": item["amount"]}

構文エラーはjson.loadsで、項目の不足・型・範囲はその後の検査で拒否されます。文字列の"120"やboolは金額として受理しません。正規化は対象の文字列項目へだけ適用しています。

4処理する入力一覧を用意

成功する二件と、金額の型が違う一件、JSONが壊れた一件を用意します。validは検証に成功したイベントの保存先です。

inputs =四つの入力 [raw, '{"id":"a-01","event":"ship","amount":120}', '{"id":"a-01","event":"pay","amount":"120"}', '{"id":']; valid = []受入れ先を空で初期化; print(len(inputs), valid)件数と受入れ前の状態

入力は4件、検証済みの一覧はまだ空です。次に各段階の失敗を確認しながら処理します。

5成功した入力だけを後段へ渡す

正しい二件、金額が文字列の一件、壊れたJSONの一件を順に処理します。

for index, text in enumerate(inputs, start=1): try: event = parse_event(text)変換と検査 valid.append(event)成功したものだけ保存 print(index, "accepted", event) except ValueError as error失敗を後段へ渡さない: print(index, "rejected", type(error).__name__)

1と2だけがvalidへ入り、3は値の検査、4はJSONの構文で拒否されます。json.JSONDecodeErrorはValueErrorの派生型なので、ここでは同じexceptで受け取りつつ型名で段階を表示しています。

6許可する状態遷移を定義する

pendingでpayを受けるとpaid、paidでshipを受けるとshippedへ進む関数を作ります。

def apply_event(states, event): transitions =許可する状態とイベントの組 {("pending", "pay"): "paid", ("paid", "ship"): "shipped"} identity = event["id"] current = states[identity]対象の現在状態 key = (current, event["event"])適用する遷移を探す if key not in transitions: raise ValueError("transition not allowed") states[identity] = transitions[key]許可した遷移だけ反映 return states[identity]

現在状態とイベントの組で、次状態を決めます。禁止した組では代入より前に例外となるので、状態を変更しません。この処理は状態遷移だけを扱い、金額の加算や残高の更新は行いません。

7入金イベントを適用

a-01はpendingから開始します。検証済みの一件目を適用します。

states = {"a-01": "pending"}初期状態; print(apply_event(states, valid[0])検証済みの入金を適用, states更新後の状態)

正規化されたIDで同じ注文を見つけ、paidへ進みます。最初の全角表記のまま辞書のキーを探していた場合とは、照合される値が異なります。

8出荷イベントを適用

二件目を同じ注文へ適用します。

print(apply_event(states, valid[1])入金済みの注文を出荷, states適用後の状態)

paidとshipの組なのでshippedになります。同じshipでも、pendingからの直接出荷や、出荷済みへの再適用は遷移表にありません。

9許可しない遷移で元の状態を保つ

出荷イベントをもう一度適用します。

try: apply_event(states, valid[1])出荷済みへ再度出荷を要求 except ValueError as error禁止した遷移: print(str(error), states拒否後に残る状態)

transition not allowedとなり、shippedのままです。同じイベントを繰り返してよいかは仕様次第で、この状態機械では再出荷を拒否しています。

10一件ごとに反映する処理

複数のイベントを順に適用し、成功するたびに状態を表示する関数です。

def apply_sequential(states, events): for event in events:一件ずつ処理 apply_event(states, event)直接状態へ反映 print("applied", states)成功した結果を公開

途中で例外になっても、それまでに反映した状態は自動では戻りません。逐次処理で途中成功を公開するという契約です。

11全件成功時に返す処理

元の辞書のコピーへ適用し、全件成功したときだけ新しい状態を返す関数です。

def apply_batch(states, events): working = states.copy()作業用の状態を分離 for event in events: apply_event(working, event)コピーへだけ反映 return working全件成功後に返す

この試料の状態値は文字列なので、辞書の浅いコピーで更新を分離できます。入れ子の可変データや外部への送信があれば、同じ方法だけで取り消せるとは限りません。

12比較する二つの状態を用意

逐次処理と全件成功時の反映を、同じpendingから始めます。二つの辞書は別々に作ります。

sequence_state = {"a-01": "pending"}逐次処理の初期状態; batch_state = {"a-01": "pending"}バッチ処理の初期状態; print(sequence_state, batch_state)

どちらも未入金の状態です。入金・出荷・再出荷という同じ三件を適用し、最後に失敗したときの状態を比較します。

13途中失敗後に残る逐次結果

pendingから入金・出荷・再出荷を実行します。

try: apply_sequential一件ごとに反映(sequence_state, [valid[0], valid[1], valid[1]]) except ValueError: print("failed", sequence_state)失敗後の公開状態

入金と出荷の二件は反映済みなので、最後の再出荷で失敗してもshippedが残ります。

14途中失敗したバッチの結果

同じイベント列を、全件成功時にだけ状態を受け取る形で処理します。

try: batch_state = apply_batch成功時だけ置き換える(batch_state, [valid[0], valid[1], valid[1]]) except ValueError: print("failed", batch_state)失敗時は元の状態

関数が正常に戻る前に失敗するので、代入は行われずpendingが残ります。同じ入力でも、途中成功を公開するか、全件成功を待つかで失敗後の状態が変わります。

まとめ

構文の解析、値の検証、項目ごとの正規化を分け、成功した入力だけを更新へ渡します。

状態機械は現在状態とイベントの組で遷移を決め、禁止した遷移では更新前に拒否します。

逐次処理とバッチ処理では、途中失敗のとき何を公開状態として残すかを決めます。

操作履歴と解説を振り返れます。

Python 3

1
import json, unicodedata; raw = '{"id":" Ａ-01 ","event":" PAY ","amount":120}'; parsed = json.loads(raw); print(parsed)
{'id': ' Ａ-01 ', 'event': ' PAY ', 'amount': 120}
2
normalized = unicodedata.normalize("NFKC", parsed["id"]).strip().casefold(); print(repr(parsed["id"]), normalized, normalized == "a-01")
' Ａ-01 ' a-01 True
3
def parse_event(text): item = json.loads(text) if not isinstance(item, dict): raise ValueError("object required") if not isinstance(item.get("id"), str) or not isinstance(item.get("event"), str): raise ValueError("id and event must be strings") if type(item.get("amount")) is not int or item["amount"] < 0: raise ValueError("amount must be a nonnegative integer") identity = unicodedata.normalize("NFKC", item["id"]).strip().casefold() event = item["event"].strip().casefold() if not identity or event not in ("pay", "ship"): raise ValueError("invalid id or event") return {"id": identity, "event": event, "amount": item["amount"]}
4
inputs = [raw, '{"id":"a-01","event":"ship","amount":120}', '{"id":"a-01","event":"pay","amount":"120"}', '{"id":']; valid = []; print(len(inputs), valid)
4 []
5
for index, text in enumerate(inputs, start=1):
    try:
        event = parse_event(text)
        valid.append(event)
        print(index, "accepted", event)
    except ValueError as error:
        print(index, "rejected", type(error).__name__)
1 accepted {'id': 'a-01', 'event': 'pay', 'amount': 120}
2 accepted {'id': 'a-01', 'event': 'ship', 'amount': 120}
3 rejected ValueError
4 rejected JSONDecodeError
6
def apply_event(states, event):
    transitions = {("pending", "pay"): "paid", ("paid", "ship"): "shipped"}
    identity = event["id"]
    current = states[identity]
    key = (current, event["event"])
    if key not in transitions:
        raise ValueError("transition not allowed")
    states[identity] = transitions[key]
    return states[identity]
7
states = {"a-01": "pending"}; print(apply_event(states, valid[0]), states)
paid {'a-01': 'paid'}
8
print(apply_event(states, valid[1]), states)
shipped {'a-01': 'shipped'}
9
try:
    apply_event(states, valid[1])
except ValueError as error:
    print(str(error), states)
transition not allowed {'a-01': 'shipped'}
10
def apply_sequential(states, events):
    for event in events:
        apply_event(states, event)
        print("applied", states)
11
def apply_batch(states, events):
    working = states.copy()
    for event in events:
        apply_event(working, event)
    return working
12
sequence_state = {"a-01": "pending"}; batch_state = {"a-01": "pending"}; print(sequence_state, batch_state)
{'a-01': 'pending'} {'a-01': 'pending'}
13
try:
    apply_sequential(sequence_state, [valid[0], valid[1], valid[1]])
except ValueError:
    print("failed", sequence_state)
applied {'a-01': 'paid'}
applied {'a-01': 'shipped'}
failed {'a-01': 'shipped'}
14
try:
    batch_state = apply_batch(batch_state, [valid[0], valid[1], valid[1]])
except ValueError:
    print("failed", batch_state)
failed {'a-01': 'pending'}
演習完了