import time
import requests


class PaymentGatewayError(Exception):
    pass


def retry(max_attempts: int = 3, sleep_sec: float = 1.0):
    def wrapper(fn):
        def inner(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return fn(*args, **kwargs)
                except Exception:
                    if attempt == max_attempts:
                        raise
                    time.sleep(sleep_sec * attempt)
        return inner
    return wrapper


@retry(max_attempts=3)
def charge_order(order_id: str, amount: int, db):
    response = requests.post(
        "https://payment.example.com/charge",
        json={"order_id": order_id, "amount": amount},
        timeout=3,
    )
@Reviewer
この時点で外部副作用は発生しています。タイムアウトや接続断の扱い次第では、アプリ側が失敗と見なしても外部側では成功済み、という状態が起こりえます。
    if response.status_code != 200:
        raise PaymentGatewayError("payment failed")
@Reviewer
status_codeだけで失敗/成功を二値化していますが、レビューでは「成功」「失敗」「不明」の3状態で読めるかを確認したいところです。
    db.execute(
        "UPDATE orders SET status = %s, charged_amount = %s WHERE id = %s",
        ("paid", amount, order_id),
    )
    db.commit()
@Reviewer
外部課金と内部状態更新を同じ再試行単位に入れているため、関数全体にretryをかける設計だと重複課金と状態不整合の両方を招きやすくなります。

@Reviewer: retry対象の単位が大きすぎます。外部課金・DB更新・再試行制御が同一責務に混在しており、結果不明な副作用を含んだまま再送しています。再試行可能な処理単位の切り分けが必要です。

@Reviewer: 外部課金APIに重複防止キーが渡されていません。同一注文の再送が区別不能なため、通信失敗時に二重課金を防げません。冪等化戦略が確認できるまで、このリトライはマージ不可です。

@Reviewer: タイムアウト時の状態が「失敗確定」として扱われていますが、外部側で処理済みの可能性があります。不明結果を failure に丸める設計だと再送判断を誤るため、unknown 状態を明示した状態遷移に見直してください。

import requests
from contextlib import contextmanager


class PaymentResultUnknown(Exception):
    pass


@contextmanager
def transaction(db):
    try:
        yield
        db.commit()
    except Exception:
        db.rollback()
        raise


def create_payment_attempt(db, order_id: str, request_id: str):
    with transaction(db):
        db.execute(
            """
            INSERT INTO payment_attempts(order_id, request_id, status)
            VALUES (%s, %s, %s)
            """,
            (order_id, request_id, "processing"),
        )
@Reviewer
先に支払い試行を記録しておくと、あとで「このrequest_idで何が起きたのか」を追跡しやすくなります。

def request_payment(order_id: str, amount: int, request_id: str):
    try:
        response = requests.post(
            "https://payment.example.com/charge",
            json={"order_id": order_id, "amount": amount},
            headers={"Idempotency-Key": request_id},
            timeout=3,
        )
    except requests.Timeout as exc:
        raise PaymentResultUnknown from exc
@Reviewer
タイムアウトを即 failure にせず、unknown として持ち上げている点が重要です。ここで二値化しないことが、この設計の出発点になります。
    if response.status_code == 200:
        return "succeeded"
    if response.status_code == 409:
        return "already_processed"
    if response.status_code == 422:
        return "rejected"
    raise RuntimeError("gateway error")


def mark_attempt_unknown(db, request_id: str):
    with transaction(db):
        db.execute(
            "UPDATE payment_attempts SET status = %s WHERE request_id = %s",
            ("unknown", request_id),
        )
@Reviewer
unknown を記録するだけでは不十分ですが、少なくとも「結果不明だった」という事実を消さない構造にはなっています。

def finalize_payment_success(db, order_id: str, amount: int, request_id: str):
    with transaction(db):
        db.execute(
            "UPDATE orders SET status = %s, charged_amount = %s WHERE id = %s",
            ("paid", amount, order_id),
        )
        db.execute(
            "UPDATE payment_attempts SET status = %s WHERE request_id = %s",
            ("succeeded", request_id),
        )
@Reviewer
orders と payment_attempts を同じトランザクションで更新しているので、片方だけ進んで状態がずれるリスクを下げられます。

def reconcile_unknown_payment(db, gateway_client, order_id: str, amount: int, request_id: str):
    result = gateway_client.fetch_payment_result(request_id)
@Reviewer
記事としてはここが一番大事です。unknown の後続確定処理が存在しないなら、「unknown を記録しただけ」で設計が止まってしまいます。
    if result == "succeeded":
        finalize_payment_success(db, order_id, amount, request_id)
        return

    if result == "not_found":
        with transaction(db):
            db.execute(
                "UPDATE payment_attempts SET status = %s WHERE request_id = %s",
                ("retryable", request_id),
            )
        return

    with transaction(db):
        db.execute(
            "UPDATE payment_attempts SET status = %s WHERE request_id = %s",
            ("failed", request_id),
        )


def charge_order(db, gateway_client, order_id: str, amount: int, request_id: str):
    create_payment_attempt(db, order_id, request_id)

    try:
        result = request_payment(order_id, amount, request_id)
    except PaymentResultUnknown:
        mark_attempt_unknown(db, request_id)
        raise

    if result in ("succeeded", "already_processed"):
        finalize_payment_success(db, order_id, amount, request_id)
        return

    with transaction(db):
        db.execute(
            "UPDATE payment_attempts SET status = %s WHERE request_id = %s",
            ("failed", request_id),
        )
@Reviewer
ここでも重要なのは「リトライ回数」ではなく、request_id 単位で結果を閉じる責務がどこにあるかです。

@Developer: でも業務上、このAPIは冪等性キーに対応していません。
@Reviewer: その場合は同期再送を前提にした設計自体が成立しません。retry回数の調整では解決できないので、この操作をリトライ対象外にするか、結果照会可能な別経路を用意する必要があります。

@Developer: unknown状態を持つのは複雑すぎませんか。
@Reviewer: 外部副作用の結果不明を failure に丸めるほうが危険です。複雑さが増えるのは unknown 状態のせいではなく、もともと外部副作用が不明結果を持つからです。複雑さを隠すより、状態として明示したほうが安全です。

@Developer: 照会APIがないので、今はretryで逃がしたいです。
@Reviewer: 照会できないなら、なおさら同期再送は危険です。再送でしか回復できない設計ではなく、保留・手動確認・後続ジョブなど別の運用経路へ切り替えるべきです。