get_event_loop→asyncio.run｜26/10

ChatGPTやClaude Codeに書かせたaiohttpの並列スクレイピングが、Python 3.14にした途端 asyncio.get_event_loop() の行で RuntimeError になって止まる。この相談が増えています。生成AIは学習時点の古い書き方を出すことがあり、エラーを貼って「直して」と頼んでも、別の古い回避策が返ってくることがあります。この記事では、AIへの指示文を実物で示します。そのうえで asyncio.run() と Semaphore を使った今の書き方に直させ、systemd timerで毎日止めずに回すところまでを順に説明します。

Python 3.14で asyncio.get_event_loop() が通らなくなった：今は asyncio.run() 1か所で始める

以前の書き方と、今の書き方

以前のasyncioのサンプルは、イベントループ（非同期処理を順番に回す仕組み）を asyncio.get_event_loop() で取得し、run_until_complete() で回す書き方が定番でした。ループが無ければ自動で作ってくれたので、スクリプトの末尾に書くだけで動いていました。

Python 3.13までは、ループが無い状態で呼ぶと警告が出るだけで動いていました。Python 3.14からはループを自動で作らなくなり、次の例外で止まります。

RuntimeError: There is no current event loop in thread 'MainThread'.

今は、非同期処理の入口を asyncio.run(main()) の1か所だけにします。asyncio.run() はループの作成・実行・後片付けまでをまとめて行うので、自分でループを扱う必要がありません。

	以前の書き方（AIが出しがち）	今の書き方
開始	`loop = asyncio.get_event_loop()` `loop.run_until_complete(main())`	`asyncio.run(main())`
Python 3.14での結果	ループが無いと `RuntimeError`	そのまま動く
後片付け	`loop.close()` を自分で書く（忘れやすい）	自動

AIがよく出す「応急処置」も採用しない

エラーを貼って直させると、AIは asyncio.new_event_loop() と asyncio.set_event_loop() でループを手作りするコードを返すことがあります。これでも動きはします。しかし、古い書き方を延命しているだけで、後片付けの漏れも残ります。指示の段階で「new_event_loop で回避しない」と明記しておくのが確実です。

実行環境

Python 3.14（手元の版は python3 -V で確認）
pip install aiohttp（入った版は pip show aiohttp で確認。オプションの細部は版によって違うので、公式ドキュメントで照合する）
urllib.robotparser は標準ライブラリなので追加インストール不要

AIに書かせる前に、python3 -V と pip show aiohttp の結果を伝えておきましょう。それだけで古い書き方が出る確率が下がります。Claude Codeのようにコマンドを実行できるエージェントなら、「まず python3 -V を実行してから書いて」と頼めます。

ChatGPT・Claude Codeへの指示文の実物：最初の指示→問題点→直させる指示→直った後のコード

最初の指示（よくある頼み方）

Pythonで https://example.com/items?page=1 から page=100 までのHTMLを
並列で取得して保存するスクリプトを書いて。なるべく速く終わるようにして。

「なるべく速く」と書くと、AIは同時接続を増やす方向に寄せます。取得先にとっては、短時間に大量のアクセスが来る書き方です。

出てきたコード（典型例）

import asyncio
import aiohttp

URLS = [f"https://example.com/items?page={i}" for i in range(1, 101)]
WEBHOOK = "https://hooks.example.com/services/T000/B000/xxxxxxxx"

async def fetch(session, url):
    try:
        async with session.get(url) as resp:
            return await resp.text()
    except Exception:
        pass

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, u) for u in URLS]
        results = await asyncio.gather(*tasks)
    # 結果を保存して WEBHOOK に通知……

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

このコードには4つの問題があります。

古いAPI：最終行の get_event_loop() がPython 3.14で RuntimeError になる
アクセス間隔が無い：gather は渡した100件を一斉に開始する。同時接続の上限は aiohttp の接続設定まかせで、待ち時間も無い
例外で黙って止まる：except Exception: pass は失敗を None にして飲み込む。全件失敗しても終了コード0で「正常終了」に見える
秘密情報の直書き：通知先のWebhook URLがコードに入っている。共有やバックアップのときに一緒に外へ出る

直させる指示（そのまま使える形）

次の条件で書き直してください。
1. 実行環境は Python 3.14。asyncio.get_event_loop() と run_until_complete は使わず、
   入口は asyncio.run(main()) の1か所だけにする。new_event_loop で回避する書き方もしない。
2. 同じサイトへの同時接続は asyncio.Semaphore で2本までにし、1リクエストごとに1秒以上空ける。
   robots.txt に Crawl-delay があれば、長いほうを使う。
3. 取得前に urllib.robotparser で robots.txt を確認し、禁止されたURLは取得しない。
4. except Exception: pass は使わない。通信エラー・タイムアウト・429・5xxだけを
   最大3回まで間隔を空けて再試行し、それ以外は失敗として記録する。
5. 1件でも失敗したら終了コード1で終わる（systemdで失敗を検知するため）。
6. URL・トークン・Webhookなどの秘密情報をコードに書かない。必要なら環境変数から読む。
7. print ではなく logging で、開始・件数・失敗したURLを出す。
変更した箇所ごとに、なぜ変えたかを1行で説明してください。

最後の1行が大事です。理由を書かせると、AIが条件を読み飛ばした箇所に気づきやすくなります。

直った後のコード

import asyncio
import logging
import sys
from datetime import date
from pathlib import Path
from urllib import robotparser

import aiohttp

BASE = "https://example.com"
URLS = [f"{BASE}/items?page={i}" for i in range(1, 11)]
USER_AGENT = "example-collector/1.0 (+https://example.com/contact)"
MAX_CONCURRENCY = 2      # 同じサイトへの同時接続数
MIN_DELAY = 1.0          # 1リクエストごとに空ける秒数
OUT_DIR = Path(__file__).resolve().parent / "data" / date.today().isoformat()

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("scraper")


def check_robots(urls):
    rp = robotparser.RobotFileParser(f"{BASE}/robots.txt")
    rp.read()
    allowed = [u for u in urls if rp.can_fetch(USER_AGENT, u)]
    delay = max(MIN_DELAY, rp.crawl_delay(USER_AGENT) or 0)
    return allowed, delay


async def fetch(session, sem, url, delay):
    async with sem:
        for attempt in range(1, 4):
            try:
                async with session.get(url) as resp:
                    resp.raise_for_status()
                    text = await resp.text()
                await asyncio.sleep(delay)
                return text
            except aiohttp.ClientResponseError as e:
                if e.status != 429 and e.status < 500:
                    raise
                log.warning("HTTP %s %s（%d回目）", e.status, url, attempt)
            except (aiohttp.ClientConnectionError, TimeoutError) as e:
                log.warning("通信エラー %s %s（%d回目）", type(e).__name__, url, attempt)
            if attempt < 3:
                await asyncio.sleep(delay * 10 * attempt)
        raise RuntimeError(f"3回失敗: {url}")


async def main(urls, delay):
    OUT_DIR.mkdir(parents=True, exist_ok=True)
    sem = asyncio.Semaphore(MAX_CONCURRENCY)
    timeout = aiohttp.ClientTimeout(total=30)
    headers = {"User-Agent": USER_AGENT}
    async with aiohttp.ClientSession(timeout=timeout, headers=headers) as session:
        results = await asyncio.gather(
            *(fetch(session, sem, u, delay) for u in urls),
            return_exceptions=True,
        )
    failed = 0
    for i, (url, result) in enumerate(zip(urls, results), start=1):
        if isinstance(result, Exception):
            failed += 1
            log.error("取得失敗 %s: %r", url, result)
            continue
        (OUT_DIR / f"{i:03}.html").write_text(result, encoding="utf-8")
    log.info("完了 成功%d件 / 失敗%d件", len(urls) - failed, failed)
    return 1 if failed else 0


if __name__ == "__main__":
    urls, delay = check_robots(URLS)
    if not urls:
        log.error("robots.txtで全URLが禁止されています")
        sys.exit(1)
    log.info("対象%d件（robots.txtで除外%d件）間隔%.1f秒", len(urls), len(URLS) - len(urls), delay)
    sys.exit(asyncio.run(main(urls, delay)))

ブロックごとの説明

定数のブロック：同時接続数（2）と間隔（1秒）を冒頭にまとめています。調整したいときにコードの中を探さずに済みます。User-Agentには連絡先URLを入れ、取得先が誰のアクセスか分かるようにしています。
check_robots()：非同期処理に入る前に、robots.txtを1回だけ読みます。禁止されたURLを除外し、Crawl-delay の指定があれば1秒と比べて長いほうを採用します。robots.txtの取得自体が通信エラーになった場合は、例外でそのまま止まり、終了コードが0以外になります。
fetch()：async with sem: の中に入れるのは同時に2本までです。1件取り終えるたびに delay 秒待ちます。再試行するのは429・5xx・通信エラー・タイムアウトだけで、待ち時間は回数に応じて延ばします。404のような「何度やっても同じ」エラーは再試行せず、すぐ失敗にします。
main()：return_exceptions=True を付けているので、1件が失敗しても残りは止まりません。失敗は握りつぶさずに件数を数え、1件でもあれば戻り値を1にします。
最終ブロック：入口は asyncio.run() の1か所だけです。戻り値を sys.exit() に渡すことで、systemdやcronが失敗を検知できるようにしています。

AIが出したコードで自分で確かめる4か所：grepで探せる形にする

コードを読み慣れていなくても、次の文字列を検索すれば危ない箇所に当たりをつけられます。

grep -nE "get_event_loop|run_until_complete|new_event_loop|except Exception|except:|Semaphore|sleep|key|token|secret|hooks" scraper.py

確認する箇所	見つけ方	あるべき状態
古いAPI	`get_event_loop`・`run_until_complete`・`new_event_loop` がヒットする	ヒットしない。入口は `asyncio.run(` の1か所だけ
アクセス間隔	`gather` があるのに `Semaphore` と `sleep` が無い	同時接続数と待ち秒数が定数として冒頭にある
例外の握りつぶし	`except Exception:` や `except:` の直後が `pass` か `return None`	捕まえる例外を絞り、失敗をログに残し、最後に終了コード1を返す
秘密情報の直書き	`key`・`token`・`secret`・`hooks` を含む文字列が値ごと書かれている	`os.environ["..."]` で読み、値はコードの外に置く

見つかったら、その行を貼ったうえで「この行は上の4項目のどれに当たるか、どう直すか」と聞き返します。AIにレビュー役をさせる指示も用意しておくと便利です。

このコードについて、(1) Python 3.14で動かない書き方、(2) アクセス間隔や同時接続数の制限が無い箇所、
(3) 例外を握りつぶしている箇所、(4) 秘密情報がコードに書かれている箇所を、行番号つきで列挙してください。
修正はまだしないでください。

「修正はまだしない」と付けるのは、指摘と修正を分けて、何が変わったかを自分で追えるようにするためです。総務省・経済産業省の「AI事業者ガイドライン（第1.2版）」（2026年3月31日公表）も、外部に影響を及ぼす操作の前に人間が最終判断する仕組みを求めています。取得先のサーバーへ毎日アクセスするスクリプトは、まさに外部に影響する操作です。定期実行に載せる前に、少なくとも上の4か所は自分の目で確認しましょう。

pandasで集計まで書かせる場合も、同じ種類の古いAPIが紛れ込みます。append削除→concat追記｜AIコード2026年10月で確認箇所をまとめています。

取得先への配慮：robots.txt・利用規約・同時接続2本と1秒間隔

利用規約を先に読む：自動取得を禁止しているサイトは、技術的に取れても対象にしません。公式APIや配布データがあれば、そちらを優先します。
robots.txtに従う：上のコードは取得前に必ず確認し、禁止されたURLを除外します。Crawl-delay があれば、その間隔を守ります。
同時接続と間隔を絞る：同時接続2本・1秒以上の間隔は「最低限」の目安です。相手のサーバーの規模が分からないなら、さらに減らしてかまいません。1日1回の収集なら、速さはほとんど意味を持ちません。
429や503が返ったら引く：「多すぎる」「今は無理」という合図なので、待ち時間を延ばして再試行し、3回でやめます。
書かせないこと：IPを変えて制限をすり抜ける方法、ログインや認証を回避する方法、取得先の制限を突破する方法はAIに頼まないでください。ブロックされたら「取得をやめる・頻度を下げる・許可を取る」のどれかです。

systemd timerで毎日回す：失敗を終了コードで検知し、どこで動かすかを決める

ユニットファイル

/home/username/.config/systemd/user/ に2つのファイルを置きます。scraper.service が「何を実行するか」、scraper.timer が「いつ実行するか」を表します。

# scraper.service
[Unit]
Description=example.com daily scraper
OnFailure=notify-failure@%n.service

[Service]
Type=oneshot
WorkingDirectory=/home/username/scraper
EnvironmentFile=/home/username/scraper/.env
ExecStart=/home/username/scraper/.venv/bin/python scraper.py
TimeoutStartSec=30min

# scraper.timer
[Unit]
Description=run scraper daily

[Timer]
OnCalendar=*-*-* 06:30:00
RandomizedDelaySec=10min
Persistent=true

[Install]
WantedBy=timers.target

ExecStart：仮想環境（venv）のPythonを直接指定します。OSの更新で python3 の版が変わっても、スクリプトが使う版は変わりません。venvは python3.14 -m venv .venv で作ります。
EnvironmentFile：通知先URLなどの秘密情報はここに書き、chmod 600 .env で本人だけが読めるようにします。コードには値を書きません。
OnFailure：スクリプトが終了コード1を返すと、通知用のユニットが起動します。通知の作り方は systemd OnFailureでSlack通知｜26年10月にまとめています。スクリプト側では、終了コードを正しく返すことだけに集中できます。
Persistent=true：停止していた間に予定時刻を過ぎた分は、次の起動時に1回実行されます。
TimeoutStartSec：通信が固まっても30分で打ち切り、失敗として扱います。

有効化と確認

systemctl --user daemon-reload
systemctl --user start scraper.service      # まず手動で1回動かす
journalctl --user -u scraper.service -n 50  # 「完了 成功N件 / 失敗N件」を確認
systemctl --user enable --now scraper.timer
systemctl --user list-timers                # 次回の実行時刻が出ればOK
sudo loginctl enable-linger username        # ログアウト中もtimerを動かす

最後の enable-linger を忘れると、ログアウトした時点でuser timerも止まります。エラーも出ないので気づきにくい点に注意してください。

どこで動かすか：止まる条件で選ぶ

置き場所	向いている条件	止まる条件
手元のPC	試作段階。数日動かして取得結果を確かめたいとき	スリープ・電源オフ・ログアウト。閉じている間は実行されない。OSの更新でPythonの版が変わる
共有レンタルサーバーのcron	1日1回・数分で終わる・追加ソフトが要らない収集	実行時間やプロセス数の制限で途中で打ち切られる。使えるPythonの版を選べず、3.14が無いことがある。`pip install` や外部への定期アクセスが規約で制限されている場合がある（事業者ごとに確認）。systemdは使えないことが多いのでcronで組む
VPS	Pythonの版を自分で固定したい・ヘッドレスブラウザを使う・実行が長い・失敗を確実に通知したい	契約や支払いの期限切れ。ディスク満杯。OSの更新でvenvの土台が変わる。監視は自分で組まないと誰も気づかない

判断の目安は次のとおりです。まず手元で動かし、journalctl に失敗が出ないことを数日確かめます。共有サーバーで要件が満たせるならcronに移し、版の固定や通知が必要ならVPSでsystemd timerを使います。どこで動かす場合も、次の3点が「止まっても気づける」ための最低条件です。

終了コードで失敗を返す
失敗を通知する
Pythonの版をvenvで固定する

まとめ：AIに書かせたasyncioスクレイピングは「入口・制限・失敗・秘密」の4点で直させる

ChatGPTやClaude Codeが出すaiohttpの並列取得は、末尾の asyncio.get_event_loop() がPython 3.14で RuntimeError になります。今の書き方は、入口を asyncio.run(main()) の1か所にすることです。AIが new_event_loop で回避しようとしたら、指示で止めさせてください。

あわせて、次の4点を指示文に書いておくと、一度で直ったコードが返ってきやすくなります。

Semaphore で同時接続を2本程度に絞り、間隔を空ける
robots.txtを確認する
except Exception: pass をやめ、失敗したら終了コード1で終わる
秘密情報を環境変数に出す

直った後は、grepで4か所を自分でも確認します。そのうえでsystemd timerに載せ、OnFailure で失敗に気づける形にしてください。手元のPC・共有サーバーのcron・VPSのどれを選ぶかは、それぞれが止まる条件と、Pythonの版を固定できるかで決めましょう。

get_event_loop→asyncio.run｜26/10

Python 3.14で asyncio.get_event_loop() が通らなくなった：今は asyncio.run() 1か所で始める

以前の書き方と、今の書き方

AIがよく出す「応急処置」も採用しない

実行環境

ChatGPT・Claude Codeへの指示文の実物：最初の指示→問題点→直させる指示→直った後のコード

最初の指示（よくある頼み方）

出てきたコード（典型例）

直させる指示（そのまま使える形）

直った後のコード

ブロックごとの説明

AIが出したコードで自分で確かめる4か所：grepで探せる形にする

取得先への配慮：robots.txt・利用規約・同時接続2本と1秒間隔

systemd timerで毎日回す：失敗を終了コードで検知し、どこで動かすかを決める

ユニットファイル

有効化と確認

どこで動かすか：止まる条件で選ぶ

関連する選択肢

まとめ：AIに書かせたasyncioスクレイピングは「入口・制限・失敗・秘密」の4点で直させる

手を動かして詰まったところまで書いた記事

PC作業効率アップおすすめガジェット

Python 3.14で asyncio.get_event_loop() が通らなくなった：今は asyncio.run() 1か所で始める

以前の書き方と、今の書き方

AIがよく出す「応急処置」も採用しない

実行環境

ChatGPT・Claude Codeへの指示文の実物：最初の指示→問題点→直させる指示→直った後のコード

最初の指示（よくある頼み方）

出てきたコード（典型例）

直させる指示（そのまま使える形）

直った後のコード

ブロックごとの説明

AIが出したコードで自分で確かめる4か所：grepで探せる形にする

取得先への配慮：robots.txt・利用規約・同時接続2本と1秒間隔

systemd timerで毎日回す：失敗を終了コードで検知し、どこで動かすかを決める

ユニットファイル

有効化と確認

どこで動かすか：止まる条件で選ぶ

関連する選択肢

まとめ：AIに書かせたasyncioスクレイピングは「入口・制限・失敗・秘密」の4点で直させる

手を動かして詰まったところまで書いた記事

PC作業効率アップ おすすめガジェット

PC作業効率アップおすすめガジェット