【アップデート】Gemini 3.5 Flashがリリースされました

Pythonで非同期処理（非同期I/O）を使ったWebスクレイパーを実装する方法を解説します。

非同期処理を使用することで、複数のウェブページへのリクエストを同時に（非ブロッキングで）行うことができ、大量のページをスクレイピングする際の速度が劇的に向上します。

### 必要なライブラリのインストール

非同期HTTPクライアントである `aiohttp` と、HTML解析のための `BeautifulSoup4`（および高速な解析器 `lxml`）を使用します。

pip install aiohttp beautifulsoup4 lxml

---

### 実装サンプル

このコードは、スクレイピング練習用サイト（Books to Scrape）から、本のタイトルと価格を同時に（非同期で）取得するスクレイパーです。

import asyncio
import time
import aiohttp
from bs4 import BeautifulSoup

# 同時接続数を制限するためのセマフォ（ターゲットサーバーへの負荷を抑えるため）
MAX_CONCURRENT_REQUESTS = 5
semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)


async def fetch_and_parse(session, url):
    """個々のページをダウンロードして解析する非同期関数"""
    async with semaphore:  # 同時実行数を制限
        try:
            timeout = aiohttp.ClientTimeout(total=10)
            print(f"[Start] Fetching: {url}")
            async with session.get(url, timeout=timeout) as response:
                if response.status != 200:
                    print(f"[Error] Status {response.status} for {url}")
                    return []
                html = await response.text()
                print(f"[Done] Fetched: {url}")
                soup = BeautifulSoup(html, "lxml")
                books = []
                for article in soup.select("article.product_pod"):
                    title = article.h3.a["title"]
                    price = article.select_one(".price_color").text
                    books.append({"title": title, "price": price})
                return books
        except asyncio.TimeoutError:
            print(f"[Timeout] {url}")
            return []
        except Exception as e:
            print(f"[Exception] {url}: {e}")
            return []


async def main():
    base_url = "https://books.toscrape.com/catalogue/page-{}.html"
    urls = [base_url.format(i) for i in range(1, 11)]
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    async with aiohttp.ClientSession(headers=headers) as session:
        tasks = [fetch_and_parse(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        all_books = []
        for r in results:
            all_books.extend(r)
        print(f"\n--- 取得完了。合計 {len(all_books)} 冊の本が見つかりました ---")
        for i, book in enumerate(all_books[:10], 1):
            print(f"{i}: {book['title']} ({book['price']})")

if __name__ == "__main__":
    start_time = time.time()
    asyncio.run(main())
    end_time = time.time()
    print(f"\n実行時間: {end_time - start_time:.2f} 秒")

---

### コードのポイント解説

1. **`aiohttp` の使用**: 通常の `requests` ライブラリは「同期（ブロッキング）」処理であるため、
   1つのリクエストが終わるまで次のリクエストを送れません。`aiohttp` を使うことで、
   レスポンスを待つ間に次のリクエストを送信できます。

2. **`asyncio.Semaphore` による流量制限**: `MAX_CONCURRENT_REQUESTS = 5` で同時にリクエストを
   送る数を制限しています。これがないと、100ページあれば100リクエストを一瞬で送信してしまい、
   相手のサーバーをダウンさせたり、IPアドレスがブロックされたりします。実務でのスクレイピングでは必須の処理です。

3. **`asyncio.gather`**: 作成した複数の非同期タスクを並行して一斉に実行します。

4. **タイムアウトの設定 (`ClientTimeout`)**: 相手サーバーが応答しない場合にプログラムが
   無限に待機するのを防ぐため、タイムアウト処理を明示的に入れています。

=== LOW thinking ===
Google Cloud（GCP）における **Cloud Run** と **Cloud Functions** は、どちらもサーバー管理が不要な
「サーバーレス（Serverless）」サービスですが、その設計思想や最適なユースケースには明確な違いがあります。

一言で言うと、**「コンテナを動かすなら Cloud Run」、「コード（関数）を動かすなら Cloud Functions」**です。

### 1. 2つのサービスの特徴

#### ■ Cloud Run とは？（コンテナベース）
Dockerなどの**コンテナイメージ**をそのままデプロイして実行するサービスです。
- **特徴:** WebアプリケーションやWeb APIの構築に最適。
- **メリット:** 言語やライブラリの制限が一切ない。1インスタンスで複数のリクエストを同時に処理できる（高効率・低コスト）。

#### ■ Cloud Functions とは？（イベント駆動型・FaaS）
特定のイベント（ファイルのアップロード、データベースの更新、Webhookなど）をトリガーにして、
**特定の関数（コード）を実行する**サービスです。
- **特徴:** 小さな単一機能の処理（マイクロサービス、バッチ処理）に最適。
- **メリット:** コンテナを作成・管理する必要がなく、ソースコードを書くだけで即座に動かせる。

### 2. 比較表

| 項目 | Cloud Run | Cloud Functions (第2世代) |
| :--- | :--- | :--- |
| **デプロイ単位** | コンテナイメージ (Docker) | ソースコード (Zip、またはGit) |
| **対応言語** | 制限なし | サポートされた言語のみ (Node, Python, Go, Java等) |
| **主な用途** | Webアプリ、Web API、マイクロサービス | イベント連携、軽量バッチ、Webhook |
| **並行処理数** | 1インスタンスあたり最大250リクエスト | 1インスタンスあたり最大1,000リクエスト（設定による） |
| **最大実行時間** | 60分 | 60分 (HTTPイベントの場合) |

### 3. どちらを選ぶべきかの判断基準

#### Cloud Run を選ぶべきケース
1. WebアプリケーションやWeb API（REST/GraphQL）を作りたい場合
2. 言語、ランタイム、システムライブラリにこだわりがある場合
3. アクセス数が多く、コストを抑えたい場合

#### Cloud Functions を選ぶべきケース
1. GCP内のイベント（トリガー）に連動させたい場合
   - 「Cloud Storageに画像がアップロードされたら、自動でサムネイルを作成する」
   - 「Firestoreのデータが更新されたら、管理者に通知メールを送る」
2. サクッと動く簡単なスクリプトやWebhookを作りたい場合

### まとめ
- **迷ったら:** まず **Cloud Run** を検討してください。特にWebサイトやAPIサーバーを作るならCloud Run一択です。
- **特定イベントの処理:** GCPの他サービスのイベントをトリガーに動かす「接着剤」のような処理を作るなら **Cloud Functions** が最適です。

=== HIGH thinking ===
マイクロサービスアーキテクチャにおける分散トランザクション管理は、システム設計において最も複雑で
困難な課題の一つです。モノリス（単一）アプリのように単一のデータベースでACID特性を保証することが
できないため、アプローチを根本から変える必要があります。

### 1. 設計思想の転換：ACID から BASE（結果整合性）へ

マイクロサービスでは、強力な一貫性（ACID）を諦め、**結果整合性（Eventual Consistency）**を受け入れる
「**BASE**（Basically Available, Soft state, Eventual consistency）」思想にシフトすることが基本です。

### 2. 主要なデザインパターン

#### パターンA：Saga パターン（強く推奨）
Sagaは、分散トランザクションを「ローカル（各サービス内）トランザクションの連鎖」として扱うパターンです。
処理が失敗した場合、**「補償トランザクション（Compensating Transaction）」**を逆順に実行してロールバックします。

- **コレオグラフィ（振付型）:** 中央集権的なコントローラーを置かず、各サービスがイベントをpub/subして自律的に連携。
- **オーケストレーション（管弦楽型）:** 「オーケストレーター」が中央でワークフローをコントロール。

#### パターンB：Transactional Outbox パターン
DBの更新とメッセージブローカーへのイベント送信をアトミックに行うパターンです。
同一のローカルDBトランザクション内でビジネスデータとOutboxテーブルへの挿入を同時に行い、
別プロセスの「Message Relayer」がOutboxテーブルを監視してメッセージブローカーに非同期で送信します。

### 3. 実装における極めて重要なベストプラクティス

- **冪等性（Idempotency）の確保:** 同じメッセージが複数回届くことを前提に設計。一意の「冪等キー」を付与してDBでチェックしてから処理を行う。
- **2相コミット（2PC）の回避:** すべてのノードがロックを保持するため、パフォーマンスが劇的に低下しアンチパターン。
- **補償トランザクションの設計:** 単純な削除ではなく、ビジネスルールに則った「逆アクション」として設計する。

### 4. 運用・監視のベストプラクティス

- **相関ID（Trace ID）の導入:** すべてのサービス呼び出し、イベント、ログにIDを伝播させる（Jaeger、Zipkinなどの分散トレーシングツールを活用）。
- **デッドレターキュー（DLQ）の活用:** 処理できない異常なメッセージを退避させ、アラートを上げて管理者が手動リカバリーできるようにする。

### まとめ：選択の指針

1. 原則として、サービスを跨ぐトランザクションが不要な設計にする（最優先）。
2. どうしても跨ぐ場合は「Saga + Outbox」パターンを採用する。
3. すべてにおいて「冪等性」と「監視性（相関ID）」を徹底する。

フィボナッチ数列を実装する代表的な3つの方法を、Pythonのコード例とともに紹介します。
それぞれ**「処理速度」「メモリ効率」「コードの分かりやすさ」**に特徴があります。

---

### 1. ループ（反復）処理による実装
**最も実用的で推奨される方法です。**

def fibonacci_loop(n):
    if n < 0:
        raise ValueError("nは0以上の整数を指定してください。")
    if n == 0:
        return 0
    a, b = 0, 1
    for _ in range(2, n + 1):
        a, b = b, a + b
    return b

print(fibonacci_loop(10))  # 出力: 55

- **メリット:** 非常に高速で、メモリの消費も最小限（変数2つ分のみ）です。
- **計算量:** 時間計算量 O(n) / 空間計算量 O(1)

---

### 2. 再帰＋メモ化による実装

from functools import lru_cache

@lru_cache(maxsize=None)
def fibonacci_memo(n):
    if n < 0:
        raise ValueError("nは0以上の整数を指定してください。")
    if n == 0:
        return 0
    if n == 1:
        return 1
    return fibonacci_memo(n - 1) + fibonacci_memo(n - 2)

print(fibonacci_memo(10))  # 出力: 55

- **メリット:** 数学的な定義に近く、コードが非常にシンプルで直感的です。
- **デメリット:** 再帰の回数が多すぎると `RecursionError` が発生します（Pythonのデフォルトではn=1000付近で限界）。
- **計算量:** 時間計算量 O(n) / 空間計算量 O(n)

---

### 3. ジェネレータによる実装（ストリーム生成）

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

gen = fibonacci_generator()
for _ in range(10):
    print(next(gen), end=" ")
# 出力: 0 1 1 2 3 5 8 13 21 34

- **メリット:** メモリ効率が極めて高い。無限に続く数列を扱う場合でもメモリをほぼ消費しません。
- **デメリット:** 特定の第n項だけをピンポイントで求める用途には不向き。
- **計算量:** 次の値を出すごとの時間計算量 O(1) / 空間計算量 O(1)

---

### まとめ

| 実装方法 | おすすめの用途 | 長所 | 短所 |
| :--- | :--- | :--- | :--- |
| **1. ループ** | 一般的な利用（最もおすすめ） | 高速、省メモリ、制限なし | コードがやや機械的 |
| **2. メモ化再帰** | アルゴリズムの学習など | 数学的定義に近く美しいコード | メモリを消費し、大きいnでエラー |
| **3. ジェネレータ** | 順番に要素を処理したい時 | 無限数列をメモリほぼゼロで扱える | 特定の第n項だけを求めるのは遅い |

--- トークン使用量 ---
入力トークン: 14
出力トークン（思考除く）: 1166
思考トークン: 1613
合計トークン: 2793