コルーチンの定義と使い方 | コルーチン | Python本格超入門

import asyncio

# async def でコルーチン関数を定義する
async def my_coroutine():
    print("コルーチンが始まりました")
    
    # 💡 time.sleep() を使うとプログラム全体がフリーズ（ブロック）するため、
    # 非同期処理では必ず asyncio.sleep() を使い、頭に await を付けます。
    await asyncio.sleep(1)  # 1秒間、実行権を周囲に譲って非同期待機
    
    print("1秒待機後、コルーチンが終了しました")

# 【重要】コルーチンは my_coroutine() と呼び出すだけでは実行されません。
# 司令塔（イベントループ）を起動する asyncio.run() に渡すことで初めて動作します。
asyncio.run(my_coroutine())

import asyncio
import time

async def fetch_data_1():
    print("データ1の取得開始...")
    await asyncio.sleep(2)  # 2秒かかる通信を想定
    print("データ1の取得完了！")

async def fetch_data_2():
    print("データ2の取得開始...")
    await asyncio.sleep(1)  # 1秒かかる通信を想定
    print("データ2の取得完了！")

async def main():
    start_time = time.time()
    
    # 2つのコルーチンを同時にイベントループで並行実行
    await asyncio.gather(fetch_data_1(), fetch_data_2())
    
    end_time = time.time()
    print(f"全体の処理時間: {end_time - start_time:.2f} 秒")

if __name__ == '__main__':
    asyncio.run(main())

import asyncio

async def download_price(item_name):
    await asyncio.sleep(1)
    return f"{item_name}の価格: 500円"

async def main():
    # パターンA: 1つずつ結果を待って受け取る
    result_single = await download_price("リンゴ")
    print(result_single)
    
    # パターンB: 複数同時に走らせて、すべての結果をリストで回収する
    # 完了した順ではなく、引数に並べた順（バナナ、ブドウ）で結果が格納されます。
    results = await asyncio.gather(
        download_price("バナナ"),
        download_price("ブドウ")
    )
    print(results)  # ['バナナの価格: 500円', 'ブドウの価格: 500円']

asyncio.run(main())

import asyncio

async def web_request(url):
    print(f"{url} へアクセス中...")
    await asyncio.sleep(1)
    
    if url == "https://error-url.com":
        # 意図的に通信エラー（例外）を発生させる
        raise ConnectionError("サーバーへの接続に失敗しました。")
    return "正常なデータ"

async def main():
    target_urls = ["https://example.com", "https://error-url.com"]
    
    for url in target_urls:
        try:
            # awaitする箇所をtryで囲むことで、非同期内のエラーを安全にキャッチ可能
            data = await web_request(url)
            print(f"取得結果: {data}")
        except ConnectionError as e:
            print(f"【エラー検知】 {e}")

asyncio.run(main())