Python におけるリトライ抽象を再考する

公開: 2026/01/27

更新: 2026/01/27

前回記事では、オプション引数をとるデコレータを楽に定義できるデコレータである kwops_decorator を作成しました。

今回は実演がてら、Python のリトライ処理記述ライブラリ Tenacity を題材に、よく使うリトライ表現を抽象化してみます。 Tenacity は基本的に、実行対象の関数にデコレータを付けることで、その関数はリトライされるべきであることを表現するのですが、関数内でリトライ回数をカウントするためには外のスコープに変数や、クラスにインスタンス変数を置いて自分で数える必要があり、カウントのリセットなども考慮すると微妙に面倒です。そこの改善を主目的に、kwops_decorator を活用し Tenacity のデコレータをラップします。

そもそも、私はほとんど Tenacity は使っておらず、職場の後輩が Tenacity のデコレータをそのまま繰り返し使ったところ、DRY の効いていないしゃびしゃびコードを生産していることを観測したので、今回の題材と相成りました。

取り組みたい要点は次の 2 つです。

試行回数を処理側で引数として取得できる（ログやメトリクスに使える）
よく使う停止条件（回数など）をプリミティブな引数で指定・合成できる
- Tenacity のデコレータでは条件ごとに対応するオブジェクトで包んで渡す必要があり、よく使う条件については冗長に感じる

出来上がったデコレータ

import sys
import functools
import tenacity
import tenacity.stop
from collections.abc import Callable
from typing import Concatenate


@kwops_decorator
def retries[**P, T](f: Callable[Concatenate[int, P], T], /, *, n: int | None = None, **kwargs) -> Callable[P, T]:
  if n:
    if "stop" in kwargs:
      kwargs["stop"] |= tenacity.stop.stop_after_attempt(n)
    else:
      kwargs["stop"] = tenacity.stop.stop_after_attempt(n)

  @functools.wraps(f)
  def closure(*f_args: P.args, **f_kwargs: P.kwargs) -> T:
    i = -1

    @tenacity.retry(**kwargs)
    def inner() -> T:
      nonlocal i
      i += 1
      return f(i, *f_args, **f_kwargs)

    return inner()

  return closure

@kwops_decorator によって retries はキーワード引数 n: int | None = None, **kwargs を取ることができる（省略することもできる）デコレータとなります。retries がデコレート対象として受け取る関数は仮引数 f です。

この retries は、tenacity.retry(...) を「そのまま使えるようにしつつ」次の機能だけを足しています。

回数 i を引数として渡す: 対象関数を f(i, *args, **kwargs) の形にして、デコレータ側でカウントした i を第 1 引数として渡します。
キーワード引数で試行上限回数を渡せる: n が指定されていれば stop_after_attempt(n) を stop に合成します。既に stop=... が来ていれば |= で OR 結合、なければ stop=... を新規に設定します。

**kwargs をそのまま tenacity.retry(...) に渡しているので、wait=, retry=, reraise= など Tenacity のオプションは好きに使えます。

import random

@retries(n=3)
def hard_to_be_done(i):
  print(f"Trial {i}")
  if random.random() >= 0.1:
    raise Exception()
  else:
    print("DONE!")

hard_to_be_done()

Trial 0
Trial 1
DONE!

複合条件

別の停止条件と組み合わせることもできます。

回数 i が先頭に注入される関係で、実引数 "spam" は仮引数 label と対応することに注意してください。

import time

@retries(n=5, stop=tenacity.stop.stop_after_delay(10))
def impossible(i, label):
  time.sleep(5)
  print(f"Trial {label} {i}")
  raise Exception()

impossible("spam")

Trial spam 0
Trial spam 1
Traceback (most recent call last):
  ...
Exception

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  ...
tenacity.RetryError: RetryError[<Future at 0x281d2778e10 state=finished raised Exception>]

ここで n=5 と stop_after_delay(10) を両方指定しているので、実際には

最大試行回数: 5 回
最大経過時間: 10 秒

のどちらかに到達した時点で停止します。例では i=1 回目と i=2 回目の間ですでに 10 秒が経過しているので、2 回試行したところで止まっています。

回数を固定したバリアント定義

プロジェクトで同じ設定を統一して使う場合は、専用のデコレータを定義するのもよいでしょう。

import sys

@kwops_decorator
def retries28(f, /, **kwargs):
  print("28回の再試行をするデコレータだぞ!", file=sys.stderr)
  return retries(n=29, **kwargs)(f)


import random

@retries28
def must_finished_task(i):
  print(f"Trial {i}")
  if random.random() >= 0.01:
    raise Exception()
  else:
    print("DONE!")

must_finished_task()

真因の例外をそのまま投げたい

ここまでの例では、最終的に再試行が失敗した場合は Tenacity の RetryError が投げられます。元の例外をそのまま外へ投げたい場合は reraise=True を渡します。**kwargs 透過的に渡していることがこういうところで活きてきます。

@retries(n=3, reraise=True)
def f(i):
  ...

このデコレータの注意点

関数定義時に先頭で i を受けるのを忘れないようにする必要がある
**kwargs を透過的に仲介するので引数の型付けが弱くなっている
async な関数は今回は未考慮（Tenacity の retry は対応しているので注意）
f(self, ...) の形をとる bound されるメソッドについては使えない
- メソッド内部にリトライ対象関数を置けば済むので、然したる問題ではない

まとめ

kwops_decorator を使うと、オプション付きデコレータの定義がどえりゃあ簡単になる
必要な抽象（特に頻出するやつ）は自分で作成し、コードのしゃびしゃび化を防ぐべし

セクション:	Python
タグ:	Python Tenacity
カテゴリ:	Tips 実験