【Python】直近N件だけを保存する:dequeで履歴と待ち行列を作る

PythonのTopに戻る


直近N件だけの履歴にはdeque(maxlen=N)、先に入れた作業を先に取り出すFIFOにはappendとpopleftを使う。dequeは両端への追加・取り出しに向いている。ただし、上限付きのdequeが満杯になると、追加と反対側の要素が自動で落ちる。処理待ちの仕事を失いたくないキューでは、この動作をそのまま使わないようにする。

履歴と作業待ちを分けて考える

ログの表示用に直近3件だけを残したい場合、古い項目が自動的に消えることは期待した動作である。一方、未処理の仕事をためる用途で同じことが起こると、作業が実行されないまま失われる。この二つは似た入れ物に見えても、満杯時に必要な動作が違う。

dequeはcollectionsにある両端キューで、appendは右端へ追加、appendleftは左端へ追加する。popとpopleftはそれぞれの端から取り出す。まずどちらの端を古い側として使うかを決めると、処理の順序を読みやすくできる。

直近の履歴とFIFOを作る

example.py

from collections import deque

history = deque(maxlen=3)
for value in range(1, 6):
    history.append(value)
print("latest:", list(history))
assert list(history) == [3, 4, 5]
history.appendleft(2)
print("append left:", list(history))
assert list(history) == [2, 3, 4]

jobs = deque(["A", "B", "C"])
processed = []
while jobs:
    processed.append(jobs.popleft())
print("FIFO:", processed)
assert processed == ["A", "B", "C"] and len(jobs) == 0
try:
    jobs.popleft()
except IndexError:
    print("empty: IndexError")
else:
    raise AssertionError("An empty deque must reject popleft")
try:
    history.insert(1, 99)
except IndexError:
    print("bounded insert: IndexError")
else:
    raise AssertionError("A full bounded deque must reject insert")

実行結果

latest: [3, 4, 5]
append left: [2, 3, 4]
FIFO: ['A', 'B', 'C']
empty: IndexError
bounded insert: IndexError

historyへ1から5まで追加しても、残るのは最後の3件だけである。そこへ左端から2を追加すると、今度は反対側の右端が落ちる。jobsは上限を付けず、左から取り出すことで、A、B、Cの順に処理している。

満杯時の自動削除を仕様として理解する

maxlen付きdequeのappendは、満杯だからといって空くまで待つ操作ではない。新しい項目を入れ、反対側から必要な数を捨てる。この性質を利用して、監視値の直近履歴や移動する小さな窓を作ることができる。

一方、insertで途中へ挿入して上限を超える場合にはIndexErrorになる。appendとすべて同じ動作ではないので、使う操作の仕様を確認したい。また、maxlenは作成後に通常の代入で変更できる設定ではない。必要なら新しいdequeを作る。

リストとの向き不向きを比べる

リストの末尾への追加は便利だが、先頭から何度も要素を取り出すと、その後ろの要素の移動が必要になる。dequeは両端の操作に向くため、先頭から取り出し続けるFIFOや、左右へ動く履歴に適している。具体的な速度差はデータや処理内容によって変わる。

その代わり、dequeの中央付近を添字で何度も参照する用途には、リストほど向いていない。通常のスライスもそのまま使う設計ではないので、必要な場面でlistへ変換する方法などを選ぶ。すべてのリストをdequeへ置き換えるのではなく、よく行う操作で選ぼう。

空のキューと複数処理の競合に注意する

空のdequeからpopやpopleftを行うとIndexErrorになる。単一の処理で順に使うなら、while jobsとして空でない間だけ取り出せる。例では取り出した後の空状態も確認している。

ただし、複数スレッドが同じdequeを使う場合、空でないと確認してから取り出すまでの間に別のスレッドが変更する可能性がある。個別の両端操作が安全に扱われることと、複数操作をひとまとまりに実行できることは別である。待ち合わせや完了管理にはqueue.Queueなどを検討する。

要素の内容を自動で複製するわけではない

dequeへ辞書やリストを入れると、そのオブジェクトへの参照を保持する。後から同じ辞書を書き換えると、保存済みの履歴も変更後の内容に見えることがある。履歴をその時点の記録として固定したいなら、追加するときに必要なコピーを行う。

maxlenで古い要素が落ちても、別の変数が同じオブジェクトを参照していれば、そのデータまで必ず消えるわけではない。dequeが持つ参照の数を制限することと、プロセス全体のメモリを厳密に制限することは区別する必要がある。

確認では、上限未満、ちょうど上限、上限を超える追加、空からの取り出しを試す。保存すべきものが「最新の履歴」なのか「必ず処理する作業」なのかを明確にすることで、便利な自動削除を誤って使うことを防げる。

動作確認と参考資料

掲載例はLinux・CPython 3.12.14で動作確認した。OS固有のコマンドや環境ごとに変わるパスは、本文中の条件を確認して使ってほしい。

関連するTips


PythonのTopに戻る