カスタムイテレータの実装

Pythonでは__iter__と__next__メソッドを実装することで、独自のイテレータを作成できます。

class NumberRange:
    def __init__(self, start, end):
        self.value = start
        self.limit = end

    def __iter__(self):
        return self

    def __next__(self):
        if self.value > self.limit:
            raise StopIteration
        current = self.value
        self.value += 1
        return current

# 使用例
for num in NumberRange(3, 7):
    print(f"値: {num}")

ジェネレータ関数の作成

yieldキーワードを使用することで、状態を保持しながら値を逐次生成する関数が作成できます。

def fibonacci_generator(limit):
    a, b = 0, 1
    while a <= limit:
        yield a
        a, b = b, a+b

# ジェネレータの使用
for n in fibonacci_generator(100):
    print(n)

複数層のデコレータ

複数のデコレータを重ねがけして、関数の前後処理を拡張できます。

def trace(func):
    def wrapper(*args, **kwargs):
        print(f"呼び出し: {func.__name__}")
        result = func(*args, **kwargs)
        print(f"戻り値: {result}")
        return result
    return wrapper

def multiply(factor):
    def decorator(func):
        def wrapper(*args, **kwargs):
            result = func(*args, **kwargs)
            return result * factor
        return wrapper
    return decorator

@trace
@multiply(5)
def calculate_sum(a, b):
    return a + b

print(f"計算結果: {calculate_sum(3, 4)}")

非同期処理の実装

async/await構文を使用して、IOバウンドな処理を非同期で実行できます。

import asyncio

async def fetch_data(delay):
    print(f"データ取得開始（{delay}秒）")
    await asyncio.sleep(delay)
    print("データ取得完了")
    return f"データ_{delay}"

async def main():
    tasks = [fetch_data(i) for i in range(1, 4)]
    results = await asyncio.gather(*tasks)
    print(f"結果: {results}")

asyncio.run(main())

型ヒントの活用

typingモジュールを使用して、変数や関数の引数・戻り値の型を明示できます。

from typing import Dict, List, Optional, Union

def process_items(items: List[Dict[str, Union[int, float]]]) -> Optional[float]:
    if not items:
        return None
    total = sum(sum(item.values()) for item in items)
    return total / len(items)

data = [
    {"a": 10, "b": 20},
    {"x": 5, "y": 15}
]
average = process_items(data)
print(f"平均値: {average}")

メタクラスによるクラス操作

typeクラスを継承したメタクラスで、クラス作成時の処理をカスタマイズできます。

class ClassMonitor(type):
    def __init__(cls, name, bases, attrs):
        print(f"クラス作成: {name}")
        print(f"属性一覧: {attrs.keys()}")
        super().__init__(name, bases, attrs)

class DataModel(metaclass=ClassMonitor):
    def save(self):
        print("データ保存処理")

class User(DataModel):
    def __init__(self, name):
        self.name = name

動的属性の追加

typesモジュールを使用して、既存オブジェクトに動的にメソッドを追加できます。

import types

class DynamicObject:
    pass

def greeting(self, name):
    return f"こんにちは、{name}さん！"

# 動的メソッド追加
obj = DynamicObject()
obj.hello = types.MethodType(greeting, obj)
print(obj.hello("山田"))

# 動的プロパティ追加
obj.version = "1.0"
print(f"オブジェクトバージョン: {obj.version}")

カスタムコンテキストマネージャ

リソース管理用にカスタムコンテキストマネージャを実装できます。

class DatabaseConnection:
    def __enter__(self):
        print("データベース接続開始")
        self.connection = "DB接続オブジェクト"
        return self.connection

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("データベース接続終了")
        if exc_type:
            print(f"エラー発生: {exc_val}")
        return True  # 例外を抑制

with DatabaseConnection() as conn:
    print(f"接続状態: {conn}")
    raise ValueError("接続タイムアウト")

マルチプロセス処理

multiprocessingモジュールでCPU密集型タスクを並列処理できます。

from multiprocessing import Pool

def square_number(n):
    return n * n

if __name__ == "__main__":
    numbers = [1, 2, 3, 4, 5]
    with Pool(processes=2) as pool:
        results = pool.map(square_number, numbers)
    print(f"結果: {results}")

マルチスレッド処理

threadingモジュールでIO密集型タスクを並列実行できます。

import threading

def download_file(file_id):
    print(f"ファイル{file_id}のダウンロード開始")
    
def process_file(file_id):
    print(f"ファイル{file_id}の処理開始")

threads = []
for i in range(3):
    t1 = threading.Thread(target=download_file, args=(i+1,))
    t2 = threading.Thread(target=process_file, args=(i+1,))
    threads.extend([t1, t2])
    t1.start()
    t2.start()

for t in threads:
    t.join()

print("すべての処理が完了しました")