Python文法100本ノック vol.9 ~デコレ―タ・関数拡張と高階関数~

Posted at 2025-08-03

まえがき

Python100本ノックについての記事です。既存の100本ノックは幾分簡単すぎるようにも感じており、それに対するアンサー記事となります。誤りなどがあれば、ご指摘ください。今回は本番編として、イテレーター・ジェネレーターと状態保持を中心に10問扱います。

Q.85

項目	内容
概要	`LazyRange` クラスを定義せよ。このクラスは組み込み `range` のように任意の `start`, `stop`, `step` に基づいて遅延的に数値を生成するが、独自の追加機能も有する。
要件	コンストラクタは `LazyRange(stop)` または `LazyRange(start, stop[, step])` をサポートすること正方向・負方向いずれもサポートすること `__iter__`, `__next__` により遅延的に値を1つずつ生成すること `__len__` により範囲の長さを返すこと `__getitem__` によりインデックスアクセスを許可すること `@property` で辞書形式の範囲情報を提供すること
発展仕様	`step == 0` に対して `ValueError` を送出インデックスが範囲外なら `IndexError` を送出インデックスが整数以外なら `TypeError` を送出 `repr()` によって `LazyRange(start=X, stop=Y, step=Z)` という文字列を返す各操作（生成・アクセス・終了など）を `loguru` でログ出力 `range_info` プロパティで現在の構成情報（辞書）を返す
使用構文	`__iter__`, `__next__`, `StopIteration`, `__len__`, `__getitem__`, `@property`, `loguru.logger`, `f-string`, `ValueError`, `IndexError`, `TypeError`, `repr`, `条件分岐`, `イテレータ状態保持`

A.85

■ 模範解答

from loguru import logger


class LazyRange:
    def __init__(self, start, stop=None, step=1):
        # stop を省略した場合の処理（range と同様）
        if stop is None:
            start, stop = 0, start

        if step == 0:
            raise ValueError("step must not be zero")

        self._start = start
        self._stop = stop
        self._step = step

        # 有効な長さを事前計算しておく（範囲外でも max(0, ...)）
        self._length = max(0, (stop - start + (step - 1 if step > 0 else step + 1)) // step)

        logger.debug(f"[INIT] LazyRange(start={start}, stop={stop}, step={step}, length={self._length})")

    def __iter__(self):
        # イテレータ初期化（再利用のため self._index で管理）
        self._current = self._start
        return self

    def __next__(self):
        # 現在の位置が終了条件に達していれば StopIteration
        if (self._step > 0 and self._current >= self._stop) or (self._step < 0 and self._current <= self._stop):
            logger.debug(f"[STOP] Iteration complete at {self._current}")
            raise StopIteration

        value = self._current
        self._current += self._step

        logger.debug(f"[YIELD] Returning {value}, next will be {self._current}")
        return value

    def __len__(self):
        # 長さを返す（事前計算済み）
        return self._length

    def __getitem__(self, index):
        # インデックスアクセス（整数以外は TypeError）
        if not isinstance(index, int):
            raise TypeError("Index must be an integer.")

        if index < 0 or index >= len(self):
            raise IndexError(f"Index {index} out of range.")

        value = self._start + index * self._step
        logger.debug(f"[ACCESS] Index {index} → Value {value}")
        return value

    def __repr__(self):
        return f"LazyRange(start={self._start}, stop={self._stop}, step={self._step})"

    @property
    def range_info(self):
        # 開始・停止・ステップ・長さの情報を返す
        return {
            'start': self._start,
            'stop': self._stop,
            'step': self._step,
            'length': self._length
        }

実行例1：通常の正方向イテレーション

r1 = LazyRange(2, 10, 2)
print(list(r1))         # [2, 4, 6, 8]
print(r1[1])            # 4
print(len(r1))          # 4
print(r1.range_info)    # {'start': 2, 'stop': 10, 'step': 2, 'length': 4}

実行例2：負方向イテレーションとログ

r2 = LazyRange(10, 2, -2)
for val in r2:
    print(val)          # 10, 8, 6, 4

print(r2[2])            # 6
print(len(r2))          # 4
print(repr(r2))         # LazyRange(start=10, stop=2, step=-2)

実行結果2：出力ログ

DEBUG    [INIT] LazyRange(start=2, stop=10, step=2, length=4)
DEBUG    [YIELD] Returning 2, next will be 4
DEBUG    [YIELD] Returning 4, next will be 6
DEBUG    [YIELD] Returning 6, next will be 8
DEBUG    [YIELD] Returning 8, next will be 10
DEBUG    [STOP] Iteration complete at 10
DEBUG    [ACCESS] Index 1 → Value 4

■ 文法・構文まとめ

文法構文	解説
`__iter__`, `__next__`	イテレーターを定義するための標準メソッド
`StopIteration`	イテレーション終了を通知する例外
`__len__`	`len(obj)` をサポート
`__getitem__`	添字アクセスに対応
`@property`	`range_info` を属性のようにアクセス可能にする
`ValueError`	無効なステップ指定（step=0）に対する明示的なエラー
`IndexError`	範囲外アクセス時に適切な例外を送出
`TypeError`	型の間違い（非整数インデックス）に対応
`loguru.logger`	全操作の詳細なトレーシング・ログ記録
`repr`	開発・デバッグで有用な状態表現をサポート

Q.86

項目	内容
概要	任意のイテラブルから要素を取り出すカスタムイテレーター `PeekableIterator` を実装せよ。`next()` による通常取得に加えて、`peek()` によって次の要素を先読みできる。ただし、`peek()` は状態を破壊しないこと。
要件	`__init__(iterable)` で任意のイテラブルを受け取る `__iter__`, `__next__` を定義し、標準イテレータとして動作 `peek()` で次の要素を確認可能（内部状態は変更不可）イテレーション完了時に StopIteration を送出
発展仕様	`collections.deque` によるバッファリング `has_next` プロパティにより要素の有無を確認可能 `reset()` メソッドにより元の状態にリセット可能（イテラブルが再生成可能である場合）すべての操作に対して `loguru` でロギング空イテレータや壊れたイテレータへの例外処理付き
使用構文	`__iter__`, `__next__`, `peek`, `collections.deque`, `StopIteration`, `property`, `try-except`, `loguru.logger`, `reset`, `isinstance`, `callable`, `f-string`

A.86

■ 模範解答

from collections import deque
from loguru import logger

class PeekableIterator:
    def __init__(self, iterable):
        # イテラブルが再生成可能かを保持（reset用）
        self._original = iterable
        self._is_reiterable = hasattr(iterable, '__iter__') and hasattr(iterable, '__getitem__')
        self._iterator = iter(iterable)  # イテレータ本体
        self._buffer = deque()           # peek用のバッファ
        logger.debug("PeekableIterator initialized.")

    def __iter__(self):
        return self

    def __next__(self):
        try:
            # バッファにpeek済み要素があればそちらを優先して返す
            if self._buffer:
                value = self._buffer.popleft()
                logger.debug(f"Next from buffer: {value}")
                return value
            # なければ次の要素を通常取得
            value = next(self._iterator)
            logger.debug(f"Next from iterator: {value}")
            return value
        except StopIteration:
            logger.warning("StopIteration raised in __next__.")
            raise

    def peek(self):
        if not self._buffer:
            try:
                # バッファが空の場合のみ先読みして格納
                value = next(self._iterator)
                self._buffer.append(value)
                logger.debug(f"Peeked value buffered: {value}")
            except StopIteration:
                logger.warning("Peek attempted but iterator is exhausted.")
                raise StopIteration("No more elements to peek.")
        return self._buffer[0]

    @property
    def has_next(self):
        try:
            self.peek()
            return True
        except StopIteration:
            return False

    def reset(self):
        if self._is_reiterable:
            self._iterator = iter(self._original)
            self._buffer.clear()
            logger.info("Iterator reset.")
        else:
            raise RuntimeError("Original iterable is not re-iterable.")

    def __repr__(self):
        return f"<PeekableIterator buffer={list(self._buffer)}>"

実行例1：基本動作の確認

from loguru import logger

# 簡易設定（出力レベル制御）
logger.remove()
logger.add(lambda msg: print(msg, end=""), level="DEBUG")

it = PeekableIterator([10, 20, 30])

print("Peek 1:", it.peek())   # -> 10
print("Next 1:", next(it))    # -> 10
print("Peek 2:", it.peek())   # -> 20
print("Next 2:", next(it))    # -> 20
print("Has next?", it.has_next)  # -> True
print("Next 3:", next(it))    # -> 30
print("Has next?", it.has_next)  # -> False

実行例2：resetとStopIterationの確認

it = PeekableIterator(range(2))
print("Next A:", next(it))  # -> 0
print("Next B:", next(it))  # -> 1

try:
    print("Peek after end:", it.peek())  # StopIteration expected
except StopIteration as e:
    print("Caught:", e)

it.reset()
print("After reset:", next(it))  # -> 0 again

■ 文法・構文まとめ

文法要素	説明
`__iter__, __next__`	イテレータプロトコルを定義し、`for`ループなどで使えるようにする構文
`peek()`	バッファを使って次の値を見せるが、状態を破壊しない先読み機構
`deque`	`collections.deque` により高速な先頭 pop/append を実現
`property`	`has_next` をプロパティ化し、直感的な存在チェックを可能に
`reset()`	オリジナルのイテラブルから再初期化。`__getitem__` などで再イテレート可能なものだけ許可
`StopIteration`	イテレータ終端を明示的に管理
`loguru.logger`	すべての動作を明示的にログ記録。デバッグ・追跡・可視化に有効
`__repr__`	デバッグや確認用にバッファ状態などを出力

Q.87

項目	内容
概要	大規模ファイルを行単位で読み込む `PersistentFileReader` クラスを実装せよ。読み取り位置を保存し、再開時にはその位置から読み込み可能にせよ。
要件	`read_lines()` でジェネレータとして行を順に返す `save_state()` により読み取り位置をファイル保存 `load_state()` により中断位置から再開保存先パスはオプションで指定可能
発展仕様	読み取り位置は `.seek()` / `.tell()` を使って保存・復元ステートファイルが存在しない場合の graceful degradation ログ出力に `loguru` を使用読み取り位置が破損した場合の例外補足と復旧複数ファイルのステート識別用にハッシュIDを使用（例：SHA256）
使用構文	`yield`, `with open`, `tell()`, `seek()`, `contextlib.suppress`, `os.path.exists`, `loguru.logger`, `hashlib.sha256`, `property`, `try-except`, `os.makedirs`

A.87

■ 模範解答

import os
import hashlib
from contextlib import suppress
from loguru import logger

class PersistentFileReader:
    def __init__(self, filepath, state_dir=".state"):
        self.filepath = filepath
        self.state_dir = state_dir
        os.makedirs(state_dir, exist_ok=True)
        self.state_path = os.path.join(state_dir, self._hash_filename(filepath) + ".state")
        logger.debug(f"PersistentFileReader initialized with file: {self.filepath}")

    def _hash_filename(self, name):
        # 複数ファイルを区別するため、ファイルパスのSHA256をステートファイル名に
        return hashlib.sha256(name.encode()).hexdigest()

    def save_state(self, pos):
        # 現在のファイル位置をステートファイルに保存
        try:
            with open(self.state_path, 'w') as f:
                f.write(str(pos))
            logger.info(f"Read position {pos} saved to {self.state_path}")
        except Exception as e:
            logger.error(f"Failed to save state: {e}")

    def load_state(self):
        # ステートファイルがあれば位置を復元、それ以外は0
        with suppress(FileNotFoundError, ValueError):
            with open(self.state_path, 'r') as f:
                pos = int(f.read())
                logger.info(f"Loaded read position {pos} from {self.state_path}")
                return pos
        logger.warning("No valid state file found, starting from beginning.")
        return 0

    def read_lines(self, save_every=1000):
        # 指定位置からファイルを開いて読み進めるジェネレータ
        try:
            with open(self.filepath, 'r', encoding='utf-8') as f:
                pos = self.load_state()
                f.seek(pos)
                logger.debug(f"Seek to position: {pos}")

                line_count = 0
                while True:
                    line = f.readline()
                    if not line:
                        break
                    yield line.rstrip('\n')
                    line_count += 1

                    # 一定行数ごとに位置を保存
                    if line_count % save_every == 0:
                        self.save_state(f.tell())

                self.save_state(f.tell())  # 最終位置も保存
                logger.info("Finished reading. Final position saved.")
        except Exception as e:
            logger.exception(f"Error during file reading: {e}")
            raise

実行例1：基本動作の確認

# 初期化
reader = PersistentFileReader("large_data.txt")

# 前回の続きから1000行ごとに位置保存しながら読む
for line in reader.read_lines(save_every=1000):
    print(line)
    if "STOP" in line:
        break  # 任意の中断

実行例2：resetとStopIterationの確認

# 1回目：前半だけ読む
reader = PersistentFileReader("large_data.txt")
for i, line in enumerate(reader.read_lines()):
    print(line)
    if i > 5:
        break  # 強制中断

# 2回目：ステートから再開
reader = PersistentFileReader("large_data.txt")
for line in reader.read_lines():
    print(line)  # 続きから出力される

■ 文法・構文まとめ

要素	解説
`yield`	巨大なファイルをメモリに載せず、1行ずつ遅延評価で処理
`tell()` / `seek()`	現在位置の取得と復元により途中再開を可能に
`os.path.exists`	ステートファイルの存在確認
`contextlib.suppress`	`FileNotFoundError` などの例外を握りつぶし、優雅なデフォルト復旧を行う
`loguru.logger`	各フェーズ（読み取り・保存・エラー等）を詳細にトレース
`hashlib.sha256`	ファイルごとのステート管理に一意なファイルIDを用いる
`os.makedirs`	ステート保存用ディレクトリを安全に生成

Q.88

項目	内容
概要	任意のイテラブルに対して、指定サイズのスライディングウィンドウを生成する `sliding_window` 関数を定義せよ。途中で停止・再開しても状態が維持されるよう実装。
要件	`sliding_window(iterable, size, *, transform=None)` の形式で定義し、サイズ `size` ごとのスライスを順に返す。オプションで変換関数 `transform(window)` を適用可能とする。
発展仕様	collections.deque による O(1) スライド実装高階関数 `transform` でウィンドウ内演算可能（例：sum, max 等）入力検証・例外補足付き loguru によるトレース・警告出力デフォルトイテラブルにも対応（再利用可能な再イテラブル対応）
使用構文	`yield`, `collections.deque`, `len`, `append`, `popleft`, `callable`, `try-except`, `loguru.logger`, `args`, `*kwargs`, `functools.wraps`
学習目的	ジェネレータ設計、状態管理、イテラブルの反復処理、関数合成、エラー処理、ログ設計、イディオム的Python実装、汎用性の高い関数定義スタイルの理解

A.88

■ 模範解答

from collections import deque
from loguru import logger
from typing import Iterable, Callable, Generator, Optional

def sliding_window(iterable: Iterable, size: int, *, transform: Optional[Callable] = None) -> Generator:
    """
    iterable: 任意のイテラブル
    size: ウィンドウサイズ（整数、>=1）
    transform: 各ウィンドウに適用する変換関数（例: sum, max, list）
    """
    if not isinstance(size, int) or size <= 0:
        raise ValueError("Window size must be a positive integer.")

    if transform is not None and not callable(transform):
        raise TypeError("transform must be callable if specified.")

    logger.debug(f"Starting sliding_window with size={size}, transform={transform}")

    window = deque(maxlen=size)  # 固定長のスライディングウィンドウ

    try:
        for item in iterable:
            window.append(item)  # 新しい要素を追加
            logger.trace(f"Appended item: {item}; Window: {list(window)}")

            if len(window) == size:
                result = transform(window) if transform else list(window)
                logger.debug(f"Yielding window: {result}")
                yield result

    except Exception as e:
        logger.exception(f"Error in sliding_window: {e}")
        raise

実行例1：基本動作の確認

from pprint import pprint

data = [10, 20, 30, 40, 50]

for window in sliding_window(data, size=3):
    pprint(window)

実行結果1

[10, 20, 30]
[20, 30, 40]
[30, 40, 50]

実行例2：高階関数による sum ウィンドウ

for s in sliding_window(range(1, 8), size=4, transform=lambda w: sum(w)):
    print(s)

実行結果2

10  # 1+2+3+4
14  # 2+3+4+5
18  # 3+4+5+6
22  # 4+5+6+7

■ 文法・構文まとめ

構文／機能	説明
`deque(maxlen=N)`	O(1) でスライドするキュー。maxlen により古い要素が自動で捨てられる
`transform()`	高階関数によって汎用性を拡張（例：合計、平均、ソートなど任意の変換）
`logger.trace()`	ステップ単位のデバッグに適したログレベル
`yield`	ストリーム処理に適したメモリ効率のよいイテレータ生成
`try-except`	入力データに依存するため、安全に処理するための例外補足

Q.89

項目	内容
概要	複数のイテラブルを指定チャンクサイズでジップしながら逐次返すクラス `ChunkZipper` を設計せよ。全てのイテラブルを並列にチャンク処理し、要素数の不一致（長さ不揃い）には明示的に対応すること。
要件	`ChunkZipper(*iterables, size=3, strict=False)` の形式で使用サイズ `size` ごとに `tuple` のチャンク列を返す `strict=True` 時は長さ不一致を検出し例外送出、`False` なら短いものは切り捨てイテラブルは何個でも対応可能
発展仕様	`itertools.islice` によるチャンク処理全イテラブルの exhaust 管理 StopIteration を握り潰して安全に終了状態ログ出力（開始・終了・チャンク内容）再イテラブルな実装バリデーション付き
使用構文	`zip`, `islice`, `itertools`, `StopIteration`, `__iter__`, `__next__`, `try-except`, `loguru.logger`, `*args`, `collections.abc.Iterable`, `raise`

A.89

■ 模範解答

from itertools import islice
from collections.abc import Iterable
from loguru import logger

class ChunkZipper:
    def __init__(self, *iterables: Iterable, size: int = 3, strict: bool = False):
        if size <= 0:
            raise ValueError("Chunk size must be a positive integer.")
        if not all(isinstance(it, Iterable) for it in iterables):
            raise TypeError("All arguments must be iterable.")

        self._iterables = [iter(it) for it in iterables]
        self._size = size
        self._strict = strict
        logger.info(f"Initialized ChunkZipper with {len(iterables)} iterables, size={size}, strict={strict}")

    def __iter__(self):
        return self

    def __next__(self):
        chunked = []

        for idx, it in enumerate(self._iterables):
            current_chunk = list(islice(it, self._size))
            logger.trace(f"Iterator {idx}: Retrieved chunk {current_chunk}")
            chunked.append(current_chunk)

        if all(len(c) == 0 for c in chunked):
            logger.info("All iterators exhausted. Stopping iteration.")
            raise StopIteration

        if self._strict and len(set(map(len, chunked))) != 1:
            logger.error(f"Length mismatch between iterables in strict mode: {list(map(len, chunked))}")
            raise ValueError("Iterables have mismatched lengths in strict mode.")

        min_len = min(map(len, chunked))
        zipped = [tuple(c[i] for c in chunked) for i in range(min_len)]
        logger.debug(f"Yielding {len(zipped)} zipped chunks: {zipped}")

        return zipped

実行例1：基本動作（strict=False）

a = range(10)
b = range(6)
cz = ChunkZipper(a, b, size=2)

for batch in cz:
    print(batch)

実行結果1

[(0, 0), (1, 1)]
[(2, 2), (3, 3)]
[(4, 4), (5, 5)]

実行例2：厳格モード（strict=True）

a = "abcdefg"
b = [1, 2, 3, 4]
cz = ChunkZipper(a, b, size=2, strict=True)

try:
    for batch in cz:
        print(batch)
except ValueError as e:
    print(f"例外捕捉: {e}")

実行結果2

[('a', 1), ('b', 2)]
[('c', 3), ('d', 4)]
例外捕捉: Iterables have mismatched lengths in strict mode.

■ 文法・構文まとめ

構文／機能	説明
`islice(it, n)`	イテレータから最大 `n` 要素を取得するチャンク操作
`tuple(c[i] for c in ...)`	各イテラブルの `i` 番目を zip のようにまとめる
`StopIteration`	終了条件を明示的に管理
`logger.trace/debug/info`	チャンク処理過程のトレース・状態ログ
`strict` モード	長さ不一致時のバリデーション（チェックなし or 例外）

Q.90

項目	内容
概要	`next()` 呼び出しごとに指定ステップで加算し、動的な `max_value` しきい値に達すると自動的にリセットされるジェネレータクラス `AutoResetCounter` を設計せよ。
要件	`next(obj)` または `next(iter(obj))` により加算可能 `max_value` はプロパティで外部変更可能自動リセット後は 0 に戻る現在のカウント値は `count` プロパティで取得任意の加算ステップ指定
発展仕様	カスタム初期値・ステップ値の指定（例：start=5, step=2）プロパティで `max_value` を動的変更手動リセット `reset()` メソッドログ出力（開始・加算・リセット）値バリデーション付き
使用構文	`__next__`, `__iter__`, `reset()`, `@property`, `@<prop>.setter`, `loguru.logger`, `try-except`, `f-string`, `raise`, `isinstance`
学習目的	状態を保持するクラス設計、イテレータ・ジェネレータの構築、プロパティ操作、例外制御、ログトレース、カプセル化と属性制御

A.90

■ 模範解答

from loguru import logger

class AutoResetCounter:
    def __init__(self, start=0, step=1, max_value=10):
        # パラメータの妥当性チェック
        if not all(isinstance(x, int) for x in [start, step, max_value]):
            raise TypeError("start, step, and max_value must all be integers.")
        if step <= 0 or max_value <= 0:
            raise ValueError("step and max_value must be positive integers.")
        
        self._start = start
        self._step = step
        self._count = start
        self._max_value = max_value
        
        logger.info(f"Initialized AutoResetCounter(start={start}, step={step}, max_value={max_value})")

    def __iter__(self):
        return self

    def __next__(self):
        current = self._count
        self._count += self._step  # 値を加算
        logger.debug(f"Incremented: {current} + {self._step} = {self._count}")

        if self._count > self._max_value:
            logger.warning(f"Threshold exceeded: {self._count} > {self._max_value}. Resetting counter.")
            self.reset()
        return current

    def reset(self):
        logger.info(f"Counter reset from {self._count} to {self._start}")
        self._count = self._start

    @property
    def count(self):
        # 現在のカウント値を返す
        return self._count

    @property
    def max_value(self):
        return self._max_value

    @max_value.setter
    def max_value(self, new_val):
        if not isinstance(new_val, int) or new_val <= 0:
            raise ValueError("max_value must be a positive integer.")
        logger.info(f"max_value changed from {self._max_value} to {new_val}")
        self._max_value = new_val

実行例1：基本的な加算と自動リセット

c = AutoResetCounter(start=0, step=3, max_value=10)

for _ in range(6):
    print(next(c))  # 自動リセット後も継続

実行結果1

0
3
6
9
0  # 12 > max_value(10) → リセット
3

実行例2：途中で max_value を動的変更

c = AutoResetCounter(start=1, step=2, max_value=7)

for _ in range(4):
    print(next(c))

c.max_value = 5  # max_value を途中で変更

for _ in range(4):
    print(next(c))

実行結果2

1
3
5
0  # リセット（7を超えた）
2
4
0  # 再びリセット（6 > max_value 5）
2

■ 文法・構文まとめ

構文／機能	説明
`__next__()`	カスタムイテレータの主構文。内部状態の更新と条件評価を定義
`@property`	外部から安全にアクセスするインターフェースを設計
`@<prop>.setter`	プロパティに動的に値を設定し、バリデーションとロギングも可能にする
`reset()`	カウントを明示的にリセット。条件により自動リセットにも使用
`loguru.logger`	状態変化やイベントのトレース（DEBUG, INFO, WARNING）

Q.91

項目	内容
概要	任意のジェネレータ関数に対し、全体数 `total` を与えることで、要素の yield ごとに進行率をログで出力するトラッキングラッパー `track_progress` を設計せよ。
要件	デコレータ構文 `@track_progress(total=N)` で使用可能各 yield ごとに現在の進行率を loguru でログ出力（％表示）デフォルト logger は `progress` ログレベル相当イテラブルの中身はそのまま透過的に返す
発展仕様	functools.wraps によるメタデータ保持 try-except によるジェネレータ内例外の補足とログジェネレータ関数・ジェネレータ式両対応複数の track_progress デコレータ同時適用可能設計
使用構文	`yield`, `@decorator`, `wraps`, `enumerate`, `try-except`, `f-string`, `loguru.logger`, `args`, `*kwargs`, クロージャ、デコレータファクトリ
学習目的	高階関数、デコレータファクトリ設計、ジェネレータのラッピング、クロージャの状態保持、ログ設計、例外処理

A.91

■ 模範解答

from functools import wraps
from loguru import logger

def track_progress(total: int):
    """
    指定された total に基づいて yield ごとの進行率をログ出力するデコレータファクトリ
    """
    if not isinstance(total, int) or total <= 0:
        raise ValueError("total must be a positive integer")

    def decorator(gen_func):
        @wraps(gen_func)
        def wrapper(*args, **kwargs):
            try:
                gen = gen_func(*args, **kwargs)  # 元のジェネレータを取得
                for idx, item in enumerate(gen, start=1):
                    progress = (idx / total) * 100
                    logger.info(f"[{gen_func.__name__}] Progress: {progress:.2f}% ({idx}/{total})")
                    yield item
            except Exception as e:
                logger.exception(f"Exception in generator '{gen_func.__name__}': {e}")
                raise
        return wrapper
    return decorator

実行例1：ファイル行を読みつつ進捗表示

@track_progress(total=5)
def dummy_reader():
    for i in range(5):
        yield f"Line {i}"

for line in dummy_reader():
    print(line)

実行結果1

[info] [dummy_reader] Progress: 20.00% (1/5)
[info] [dummy_reader] Progress: 40.00% (2/5)
[info] [dummy_reader] Progress: 60.00% (3/5)
[info] [dummy_reader] Progress: 80.00% (4/5)
[info] [dummy_reader] Progress: 100.00% (5/5)
Line 0
Line 1
Line 2
Line 3
Line 4

実行例2：数値演算を含むジェネレータに適用

@track_progress(total=4)
def squared_gen(nums):
    for x in nums:
        yield x ** 2

for val in squared_gen([1, 2, 3, 4]):
    print(val)

実行結果2

[info] [squared_gen] Progress: 25.00% (1/4)
[info] [squared_gen] Progress: 50.00% (2/4)
[info] [squared_gen] Progress: 75.00% (3/4)
[info] [squared_gen] Progress: 100.00% (4/4)
1
4
9
16

■ 文法・構文まとめ

構文／機能	説明
`@wraps(func)`	元関数のメタデータ（名前・docstring等）を保つ
`decorator factory`	引数付きデコレータ（例：`@track_progress(total=...)`）
`enumerate()`	インデックス付きイテレーションで進捗割合を計算
`yield`	ラッパーでも元ジェネレータの出力を透過的に返す
`logger.info()`	進捗メッセージをログ出力
`try-except`	ジェネレータ内エラーを安全に補足・再送出し、トレースログ付きで表示

Q.92

項目	内容
概要	外部入力に応じてステートを内部保持し、段階的に出力を切り替える状態遷移ジェネレータ関数を定義せよ。状態は `START` → `PROCESSING` → `FINAL` を取り、入力データによって出力が変化するようにせよ。send() は使わず、状態は内部変数で管理すること。
要件	`stateful_generator(iterable)` 形式で実装初期状態 `START`、次に `PROCESSING`、最後に `FINAL` で完了各状態で出力が異なるログに状態遷移を明示
発展仕様	イテラブルの中身に応じて明示的に状態遷移する例外発生時に `ERROR` 状態へ遷移・ログステートは列挙型 Enum で管理し明示性向上 loguru による全トレース・状態変化出力最終出力は総まとめも出力
使用構文	`yield`, `for`, `while`, `if-elif`, `Enum`, `try-except`, `loguru.logger`, `state`, `list.append`, `in`, `str.upper`, `@staticmethod`
学習目的	イテレータ設計、状態保持設計、Enumによる明示的状態制御、ステートマシンパターン設計、例外ハンドリング、ログ出力制御

A.92

■ 模範解答

from enum import Enum, auto
from loguru import logger
from typing import Generator, Iterable


class State(Enum):
    START = auto()
    PROCESSING = auto()
    FINAL = auto()
    ERROR = auto()


def stateful_generator(data: Iterable[str]) -> Generator[str, None, None]:
    """
    入力に基づき状態を持って動作するステートマシン型ジェネレータ。
    START → PROCESSING → FINAL へと遷移。
    send() は使用せず、内部状態で制御。
    """
    state = State.START
    buffer = []  # 中間状態保持用

    logger.info(f"Initial state: {state.name}")

    try:
        for item in data:
            item_upper = item.strip().upper()  # 事前整形
            logger.debug(f"Received: {item_upper} | Current state: {state.name}")

            if state == State.START:
                if item_upper == "BEGIN":
                    logger.info("Transition START → PROCESSING")
                    state = State.PROCESSING
                    yield "Started processing"
                else:
                    logger.warning("Ignoring input in START state")
                    yield "Waiting to start"

            elif state == State.PROCESSING:
                if item_upper == "END":
                    logger.info("Transition PROCESSING → FINAL")
                    state = State.FINAL
                    yield f"Processed {len(buffer)} items"
                else:
                    buffer.append(item_upper)
                    logger.trace(f"Buffered: {item_upper}")
                    yield f"Processing: {item_upper}"

            elif state == State.FINAL:
                yield "Finalizing output..."
                logger.success("State machine completed.")
                break

    except Exception as e:
        logger.exception("Unexpected error during stateful generation.")
        state = State.ERROR
        yield f"Error occurred: {e}"

    finally:
        if state == State.FINAL and buffer:
            yield f"Summary: {', '.join(buffer)}"
            logger.info("Emitted summary.")

実行例1：通常の BEGIN → PROCESSING → END

data = ["ignore", "BEGIN", "alpha", "beta", "END", "after"]

for output in stateful_generator(data):
    print(output)

実行結果1

Waiting to start
Started processing
Processing: ALPHA
Processing: BETA
Processed 2 items
Finalizing output...
Summary: ALPHA, BETA

実行例2：END が来ないケース（途中終了）

data = ["BEGIN", "x", "y", "z"]

for output in stateful_generator(data):
    print(output)

実行結果2

Started processing
Processing: X
Processing: Y
Processing: Z
# 状態が FINAL に到達しないため、Summary は出ない

■ 文法・構文まとめ

構文・機能	説明
`Enum.auto()`	自動的に連番値を振る Enum クラスのメンバ定義
`state = State.X`	状態遷移の明示的管理（列挙型で安全・明確）
`yield`	ストリーミング出力として状態出力を段階的に返す
`try-except-finally`	安全なジェネレータ実行と終了処理保証
`logger.trace/debug/info`	状態とデータ処理のトレースログ出力
`str.upper/strip()`	一貫した入力整形による状態遷移判断

Q.93

項目	内容
概要	任意のイテラブルを対象とし、進行状態（インデックス）を pickle により一時ファイルへ保存し、次回起動時に自動復元する再開可能イテレーター `ResumableIterator` を定義せよ。
要件	イテレーション中に `.pkl` ファイルを用いて状態を保存次回は保存状態から自動的に再開イテラブルは `list` or `tuple` 前提（インデックスアクセス可能型）
発展仕様	一時ファイルパスは自動生成または明示指定可能例外時にも安全に状態保存（`finally`活用）進行率ログを `loguru` で表示途中で中断しても次回再開できる終了後は保存ファイルを自動削除
使用構文	`__iter__`, `__next__`, `pickle`, `tempfile`, `os.path.exists`, `try-except-finally`, `with open`, `loguru.logger`, `len`, `del`
学習目的	イテレーター設計、状態保持のファイル化、復元処理、エラー安全性、ログ活用、Python的なクラス設計パターンの理解

A.93

■ 模範解答

import os
import pickle
import tempfile
from loguru import logger
from typing import Iterable, Optional


class ResumableIterator:
    def __init__(self, iterable: Iterable, save_path: Optional[str] = None):
        self.data = list(iterable)  # 再現性のため list 化
        self.index = 0              # 初期インデックス

        # 状態保存用パス（指定されなければ一時ファイルに保存）
        self.save_path = save_path or os.path.join(tempfile.gettempdir(), "resumable_iter.pkl")

        # 保存ファイルが存在する場合は状態を復元
        if os.path.exists(self.save_path):
            try:
                with open(self.save_path, 'rb') as f:
                    self.index = pickle.load(f)
                    logger.info(f"復元された状態: index={self.index}")
            except Exception as e:
                logger.error(f"復元失敗: {e}")
                self.index = 0

        self.length = len(self.data)  # 全体長

    def __iter__(self):
        return self

    def __next__(self):
        try:
            if self.index >= self.length:
                logger.success("イテレーション完了。状態ファイルを削除します。")
                self._cleanup()
                raise StopIteration

            item = self.data[self.index]  # 現在のアイテム取得
            logger.debug(f"出力: {item} (index: {self.index})")

            self.index += 1  # 次に進む
            self._save_state()  # 状態保存

            self._log_progress()
            return item

        except Exception as e:
            logger.exception("次の要素の取得中にエラー発生")
            raise e

    def _save_state(self):
        """ 現在の index を pickle で保存 """
        try:
            with open(self.save_path, 'wb') as f:
                pickle.dump(self.index, f)
                logger.trace(f"状態を保存: index={self.index}")
        except Exception as e:
            logger.error(f"状態保存に失敗: {e}")

    def _cleanup(self):
        """ 状態ファイルを削除 """
        if os.path.exists(self.save_path):
            try:
                os.remove(self.save_path)
                logger.debug("状態ファイル削除完了")
            except Exception as e:
                logger.warning(f"状態ファイル削除失敗: {e}")

    def _log_progress(self):
        """ 進行状況ログを出力 """
        percent = (self.index / self.length) * 100
        logger.info(f"進行率: {self.index}/{self.length} ({percent:.2f}%)")

実行例1：途中まで実行して強制終了

# 初回実行（途中まで処理して中断）
it = ResumableIterator(['a', 'b', 'c', 'd'])

for i, v in enumerate(it):
    print(v)
    if i == 1:
        break  # 強制中断（次回再開に備える）

実行結果1

a
b
# 再実行時は c から再開される

実行例2：再起動後、c から再開される

# 続きから再開（状態は保存されている）
it = ResumableIterator(['a', 'b', 'c', 'd'])

for v in it:
    print(v)

実行結果2

c
d
# 完了後、状態ファイルは削除される

■ 文法・構文まとめ

構文・機能	説明
`pickle.dump/load()`	任意オブジェクトの状態をシリアライズ／デシリアライズ可能
`tempfile.gettempdir()`	システム標準の一時ファイルパス取得
`__next__`, `__iter__`	Python イテレータの基本プロトコル
`os.path.exists`, `os.remove`	ファイルの有無・削除のための OS 系操作
`try-except-finally`	ファイル復元・保存の安全な処理構造
`loguru.logger`	高機能ロガーで進行率・復元・例外処理を明示的に記録

Q.94

項目	内容
概要	任意のイテラブルに対して `map`, `filter`, `reduce` 的処理を順にパイプライン構成し、処理を遅延評価しつつ逐次適用できるイテレータクラス `FunctionalPipe` を実装せよ。
要件	`FunctionalPipe(iterable)` に対して `.map(f)`, `.filter(pred)`, `.reduce(func)` のようにチェーン呼び出し可能 reduce は遅延評価とは別に `.run()` で集計処理を実行イテレータとして next() 対応
発展仕様	状態は内部ステップリストに保持され、順次適用される関数の型検査とエラー捕捉を含める loguru による各ステップのログ出力途中キャンセル（例外）時も処理経路が追跡可能
使用構文	`__iter__`, `__next__`, `map`, `filter`, `reduce`, `lambda`, `callable`, `try-except`, `functools.reduce`, `loguru.logger`, `@property`
学習目的	高階関数によるチェーン設計、内部状態の持続処理、Pythonic なメソッドチェーン設計、エラーハンドリング、遅延評価と集計の使い分け、ログ設計

A.94

■ 模範解答

import os
import pickle
import tempfile
from loguru import logger
from typing import Iterable, Optional


class ResumableIterator:
    def __init__(self, iterable: Iterable, save_path: Optional[str] = None):
        self.data = list(iterable)  # 再現性のため list 化
        self.index = 0              # 初期インデックス

        # 状態保存用パス（指定されなければ一時ファイルに保存）
        self.save_path = save_path or os.path.join(tempfile.gettempdir(), "resumable_iter.pkl")

        # 保存ファイルが存在する場合は状態を復元
        if os.path.exists(self.save_path):
            try:
                with open(self.save_path, 'rb') as f:
                    self.index = pickle.load(f)
                    logger.info(f"復元された状態: index={self.index}")
            except Exception as e:
                logger.error(f"復元失敗: {e}")
                self.index = 0

        self.length = len(self.data)  # 全体長

    def __iter__(self):
        return self

    def __next__(self):
        try:
            if self.index >= self.length:
                logger.success("イテレーション完了。状態ファイルを削除します。")
                self._cleanup()
                raise StopIteration

            item = self.data[self.index]  # 現在のアイテム取得
            logger.debug(f"出力: {item} (index: {self.index})")

            self.index += 1  # 次に進む
            self._save_state()  # 状態保存

            self._log_progress()
            return item

        except Exception as e:
            logger.exception("次の要素の取得中にエラー発生")
            raise e

    def _save_state(self):
        """ 現在の index を pickle で保存 """
        try:
            with open(self.save_path, 'wb') as f:
                pickle.dump(self.index, f)
                logger.trace(f"状態を保存: index={self.index}")
        except Exception as e:
            logger.error(f"状態保存に失敗: {e}")

    def _cleanup(self):
        """ 状態ファイルを削除 """
        if os.path.exists(self.save_path):
            try:
                os.remove(self.save_path)
                logger.debug("状態ファイル削除完了")
            except Exception as e:
                logger.warning(f"状態ファイル削除失敗: {e}")

    def _log_progress(self):
        """ 進行状況ログを出力 """
        percent = (self.index / self.length) * 100
        logger.info(f"進行率: {self.index}/{self.length} ({percent:.2f}%)")

実行例1：途中まで実行して強制終了

# 初回実行（途中まで処理して中断）
it = ResumableIterator(['a', 'b', 'c', 'd'])

for i, v in enumerate(it):
    print(v)
    if i == 1:
        break  # 強制中断（次回再開に備える）

実行結果1

a
b
# 再実行時は c から再開される

実行例2：再起動後、c から再開される

# 続きから再開（状態は保存されている）
it = ResumableIterator(['a', 'b', 'c', 'd'])

for v in it:
    print(v)

実行結果2

c
d
# 完了後、状態ファイルは削除される

■ 文法・構文まとめ

構文・機能	説明
`pickle.dump/load()`	任意オブジェクトの状態をシリアライズ／デシリアライズ可能
`tempfile.gettempdir()`	システム標準の一時ファイルパス取得
`__next__`, `__iter__`	Python イテレータの基本プロトコル
`os.path.exists`, `os.remove`	ファイルの有無・削除のための OS 系操作
`try-except-finally`	ファイル復元・保存の安全な処理構造
`loguru.logger`	高機能ロガーで進行率・復元・例外処理を明示的に記録

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up