requestsで1ページずつ取りに行くスクリプトを、httpxの AsyncClient で並列化しようとして、手元のコードが proxies= の行で止まった方は多いはずです。httpx 0.28では proxies= 引数が削除され、proxy= か mounts= で書く形になりました。この記事では、httpx 0.28の書き方で AsyncClient・Limits・asyncio.Semaphore を組み合わせ、同時接続数とアクセス間隔を守ったまま並列取得するコードを示します。あわせて、systemd timerで毎日止めずに回す設定まで説明します。

httpx 0.28で変わった書き方：proxies= から proxy= / mounts= へ

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

社内ネットワークの出口プロキシ経由でしか外に出られない環境では、以前はクライアント生成時に proxies= へ辞書を渡していました。httpx 0.28ではこの引数が無くなっています。

# 以前の書き方（httpx 0.28 ではこの行で止まる）
import httpx

client = httpx.Client(
    proxies={
        "http://": "http://proxy.example.com:8080",
        "https://": "http://proxy.example.com:8080",
    }
)

実行すると TypeError: Client.__init__() got an unexpected keyword argument 'proxies' になります。引数そのものが無いため、警告ではなく例外で止まります。今の書き方は次の2通りです。

# 今の書き方1：全通信を同じプロキシに通すなら proxy= に文字列を1つ渡す
client = httpx.AsyncClient(proxy="http://proxy.example.com:8080")

# 今の書き方2：スキームや宛先ごとに分けたいなら mounts= にトランスポートを渡す
limits = httpx.Limits(max_connections=3, max_keepalive_connections=3)
mounts = {
    "http://": httpx.AsyncHTTPTransport(proxy="http://proxy.example.com:8080", limits=limits),
    "https://": httpx.AsyncHTTPTransport(proxy="http://proxy.example.com:8080", limits=limits),
}
client = httpx.AsyncClient(mounts=mounts)
  • 書き方1：ほとんどの用途はこれで足ります。proxies= の辞書で同じURLを2回書いていたなら、そのまま proxy= 1つに置き換えられます。
  • 書き方2：mounts= に渡したトランスポート（実際の通信を担う部品）は、クライアントに渡した limits= を引き継ぎません。同時接続数を絞りたいなら、上のように各トランスポートにも limits= を渡します。ここを忘れると、後述の接続数制限が黙って効かなくなります。

環境変数でプロキシを渡す場合

httpxは既定で HTTPS_PROXY・HTTP_PROXY・NO_PROXY の環境変数を読みます（trust_env=True）。コードにプロキシのURLを書きたくない場合は、後述のsystemdのユニットファイルで環境変数として渡せば、コード側は何も書かずに済みます。

requestsとの違いも押さえておく

項目

requests

httpx 0.28

プロキシ指定

proxies={"https": "..."}（キーはスキーム名）

proxy="..." または mounts=

非同期

非対応

httpx.AsyncClient

接続数の上限

アダプタの設定で間接的に

httpx.Limits で直接

タイムアウト既定値

指定しなければ無期限に待つ

既定で5秒（httpx.Timeout で細かく指定可）

手元のバージョンは pip show httpx で確認できます。この記事のコードは proxies= が無い0.28以降を前提にしています。

requestsの逐次取得をAsyncClient＋Limits＋Semaphoreで並列化する

書き換え前：requestsで1件ずつ取る

import time
import requests

URLS = [f"https://example.com/items?page={i}" for i in range(1, 21)]

with requests.Session() as session:
    for url in URLS:
        r = session.get(url, timeout=10)
        r.raise_for_status()
        # ...解析...
        time.sleep(1)

この書き方は安全ですが、相手の応答を待つ時間がそのまま積み上がります。仮に1ページの応答に2秒かかるなら、20ページで「(2秒＋1秒の待機) × 20 ＝ 約60秒」です。

並列化の目的は「相手への頻度を上げること」ではない

ここが一番大事な点です。並列化して得したいのは応答待ちの時間を重ねることであって、相手サーバーへのリクエスト頻度を上げることではありません。そこで次の3つを別々の役割で使います。

部品

何を制限するか

無いとどうなるか

httpx.Limits

同時に張るTCP接続の数（接続プールの上限）

タスクの数だけ接続を張りにいく

asyncio.Semaphore

同時に処理中のタスクの数（取得・待機・解析を含む）

全タスクがプールの空き待ちに並び、PoolTimeout で落ちる

自前の間隔制御

同一ホストへのリクエスト開始の間隔

同時接続数は守れても、瞬間的に連続でリクエストが飛ぶ

Limits だけで絞ると、上限を超えたタスクは接続プールの空きを待ち、その待ちが Timeout の pool を超えた時点で例外になります。数百件を一度に投げると起きやすいので、Semaphore で「そもそも同時に動くタスク」を絞っておきます。さらに Semaphore は同時実行数しか見ないため、開始間隔は別に制御します。

この形なら、間隔1秒・応答2秒・20ページでも、おおよそ「1秒 × 20 ＋ 最後の応答待ち2秒 ≒ 22秒」の計算になります（実際の時間は相手の応答次第です）。相手から見たリクエスト頻度は逐次版と変わりません。

実行環境

  • Python 3.11以降。Python 3.10は2026年10月1日でサポートが終了しました（3.10.22が最終リリース）。3.11は2027年10月まで、3.12は2028年10月までセキュリティ修正が提供されます。
  • pip install "httpx>=0.28" beautifulsoup4 pandas
  • httpxは proxies= が削除された書き方を前提にするため、>=0.28 を付けています。古い仮想環境（venv）を使い回している場合は、venv は中のPythonを差し替えられないので、新しいPythonで作り直してください。

完成形のコード（collector.py）

import asyncio
import logging
import sys
import time
from pathlib import Path
from urllib.parse import urljoin
from urllib.robotparser import RobotFileParser

import httpx
import pandas as pd
from bs4 import BeautifulSoup

BASE = "https://example.com"
URLS = [f"{BASE}/items?page={i}" for i in range(1, 21)]
USER_AGENT = "example-collector/1.0 (+https://example.com/contact)"
MAX_CONCURRENCY = 3       # 同時に動かすタスク数・接続数
MIN_INTERVAL = 1.0        # 同一ホストへのリクエスト開始間隔（秒）
MAX_RETRIES = 3
OUTPUT = Path("/home/username/scraper/data/items.csv")

log = logging.getLogger("collector")


class RateLimiter:
    """リクエストの開始時刻を、前回から interval 秒以上空ける"""

    def __init__(self, interval: float):
        self.interval = interval
        self._lock = asyncio.Lock()
        self._last = 0.0

    async def wait(self) -> None:
        async with self._lock:
            delay = self._last + self.interval - time.monotonic()
            if delay > 0:
                await asyncio.sleep(delay)
            self._last = time.monotonic()


async def load_robots(client: httpx.AsyncClient) -> RobotFileParser:
    rp = RobotFileParser()
    r = await client.get(urljoin(BASE, "/robots.txt"))
    if r.status_code == 200:
        rp.parse(r.text.splitlines())
    elif r.status_code in (401, 403):
        rp.disallow_all = True
    else:
        rp.allow_all = True
    return rp


def retry_after_seconds(response: httpx.Response, fallback: float) -> float:
    value = response.headers.get("Retry-After", "")
    if value.isdigit():
        return min(int(value), 300)
    return fallback


def parse(html: str, url: str) -> list[dict]:
    soup = BeautifulSoup(html, "html.parser")
    rows = []
    for card in soup.select("div.item"):
        name = card.select_one(".name")
        price = card.select_one(".price")
        if name is None or price is None:
            continue
        rows.append({
            "url": url,
            "name": name.get_text(strip=True),
            "price": price.get_text(strip=True),
        })
    return rows


async def fetch(client, url, sem, limiter) -> list[dict]:
    async with sem:
        for attempt in range(1, MAX_RETRIES + 1):
            await limiter.wait()
            try:
                r = await client.get(url)
            except httpx.TransportError as e:
                log.warning("通信失敗 %s（%d回目）: %s", url, attempt, e)
                await asyncio.sleep(2 ** attempt)
                continue
            if r.status_code in (429, 503):
                wait = retry_after_seconds(r, fallback=2 ** attempt * 5)
                log.warning("%d %s %.0f秒待機", r.status_code, url, wait)
                await asyncio.sleep(wait)
                continue
            r.raise_for_status()
            return parse(r.text, url)
    raise RuntimeError(f"{MAX_RETRIES}回失敗: {url}")


def save(rows: list[dict]) -> int:
    df = pd.DataFrame(rows, columns=["url", "name", "price"])
    df["price"] = pd.to_numeric(
        df["price"].str.replace(r"[^\d]", "", regex=True), errors="coerce"
    )
    df = df.drop_duplicates(subset=["url", "name"])
    df["fetched_at"] = pd.Timestamp.now(tz="Asia/Tokyo")

    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
    tmp = OUTPUT.with_suffix(".tmp")
    df.to_csv(tmp, index=False, encoding="utf-8")
    tmp.replace(OUTPUT)
    return len(df)


async def main() -> int:
    limits = httpx.Limits(
        max_connections=MAX_CONCURRENCY,
        max_keepalive_connections=MAX_CONCURRENCY,
    )
    timeout = httpx.Timeout(10.0, pool=30.0)
    async with httpx.AsyncClient(
        limits=limits,
        timeout=timeout,
        headers={"User-Agent": USER_AGENT},
        follow_redirects=True,
    ) as client:
        robots = await load_robots(client)
        targets = [u for u in URLS if robots.can_fetch(USER_AGENT, u)]
        if len(targets) < len(URLS):
            log.info("robots.txt により %d件を除外", len(URLS) - len(targets))

        interval = max(MIN_INTERVAL, robots.crawl_delay(USER_AGENT) or 0)
        limiter = RateLimiter(interval)
        sem = asyncio.Semaphore(MAX_CONCURRENCY)

        results = await asyncio.gather(
            *(fetch(client, u, sem, limiter) for u in targets),
            return_exceptions=True,
        )

    rows, failed = [], 0
    for url, res in zip(targets, results):
        if isinstance(res, BaseException):
            failed += 1
            log.error("取得できず %s: %r", url, res)
        else:
            rows.extend(res)

    saved = save(rows)
    log.info("保存 %d件 / 失敗 %dページ", saved, failed)
    return 0 if failed == 0 and saved > 0 else 1


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    sys.exit(asyncio.run(main()))

ブロックごとの説明

  1. 定数：USER_AGENT には自分の連絡先URLを入れます。ブラウザの値を名乗るのではなく、何者がアクセスしているかを相手が分かるようにする目的です。OUTPUT は例として /home/username/ 配下にしています。
  2. RateLimiter：asyncio.Lock で「前回の開始時刻」を1つだけ持ち、全タスクで共有します。3タスクが同時に wait() を呼んでも順番に通され、開始は必ず interval 秒ずつずれます。time.monotonic() を使うのは、システム時刻が変わっても間隔が狂わないようにするためです。
  3. load_robots：標準ライブラリの urllib.robotparser は自前で通信すると今回のクライアント設定（プロキシ・UA）を通らないので、robots.txtだけhttpxで取ってから parse() に渡しています。401/403なら全拒否、404などで存在しなければ全許可として扱います。
  4. retry_after_seconds：429（リクエスト過多）や503（一時停止）で相手が Retry-After を返していれば、その秒数に従います。上限300秒は、異常な値で何時間も止まらないための保険です。
  5. parse：BeautifulSoupのCSSセレクタで項目を拾います。div.item・.name・.price は例なので、対象ページに合わせて変えてください。要素が欠けたカードは飛ばし、1件の構造違いで全体を落とさないようにしています。
  6. fetch：async with sem の内側でリトライとバックオフ（失敗のたびに待ち時間を伸ばすこと）をしています。待機中もセマフォを握ったままなのは意図的で、相手が詰まっているときにこちらの同時実行数が自然に下がるようにするためです。
  7. save：pandasで価格を数値化し、重複を落とし、取得時刻を付けます。一時ファイルに書いてから replace() で置き換えるので、途中で落ちても前回のCSVが中途半端に壊れることはありません。encoding="utf-8" はPython 3.15でUTF-8が既定になった後も、他の環境で読む場合に備えて明示しています。
  8. main：Timeout(10.0, pool=30.0) は「通信は10秒、プールの空き待ちは30秒」の意味です。asyncio.TaskGroup ではなく gather(..., return_exceptions=True) を使っているのは、TaskGroupだと1ページの失敗で残りのタスクが全部キャンセルされるためです。毎日の収集では「取れた分は保存し、失敗は数えて報告する」方が扱いやすくなります。
  9. 終了コード：失敗が1ページでもあるか、保存が0件なら 1 を返します。これがsystemd側で「失敗」として記録される根拠になります。

systemd timerで毎日止めずに回す：service・timer・lingerの3点

ユニットファイル

cronとの違いや、停止中に逃した回をどう拾うかは systemd timerでcron取りこぼし対策｜26年10月 で詳しく書いたので、ここでは今回のスクリプト用の最小構成だけ示します。ユーザー単位（root不要）で置きます。

/home/username/.config/systemd/user/collector.service

[Unit]
Description=Daily collector (httpx AsyncClient)

[Service]
Type=oneshot
WorkingDirectory=/home/username/scraper
ExecStart=/home/username/scraper/.venv/bin/python collector.py
TimeoutStartSec=30min
# 出口プロキシが必要な環境だけ有効にする
# Environment=HTTPS_PROXY=http://proxy.example.com:8080

/home/username/.config/systemd/user/collector.timer

[Unit]
Description=Run collector daily

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

[Install]
WantedBy=timers.target
  • ExecStart にvenvのPythonを直接書く：source activate は不要です。venv内の python を絶対パスで呼べば、そのvenvのパッケージが使われます。
  • Type=oneshot：終わったら終了するスクリプト向けの型です。sys.exit(1) で終われば失敗として記録されます。
  • TimeoutStartSec：通信が想定外に詰まっても、30分で打ち切って次の日に影響を残しません。
  • Persistent=true：マシンが止まっていた時間帯の分を、起動後に1回実行します。
  • RandomizedDelaySec：毎日ぴったり同じ秒に集中しないよう、開始を最大10分ずらします。

有効化と確認のコマンド

systemctl --user daemon-reload
systemctl --user enable --now collector.timer
systemctl --user list-timers            # 次回の実行予定を確認
systemctl --user start collector.service # 手動で1回試す
journalctl --user -u collector.service -n 50   # ログを見る
systemctl --user list-units --failed     # 失敗したユニットを一覧
sudo loginctl enable-linger username     # ログアウト後もタイマーを動かす

最後の enable-linger を忘れると、SSHを切った時点でユーザー単位のタイマーが止まります。設定変更はこの1行だけで、元に戻すときは sudo loginctl disable-linger username です。

どこで動かすかで「止まる条件」が変わる

実行場所

向いている条件

止まる条件

手元のPC（launchd・タスクスケジューラ）

動作確認、週に数回でよい収集

スリープ・蓋を閉じる・電源を切る。起動していない日は実行されない

共有レンタルサーバーのcron

数分で終わる処理を1日1回

実行時間やプロセスの制限に当たると強制終了される。systemdは基本的に使えない。制限の具体値は各社の仕様で確認する

VPS（systemd timer）

毎日確実に回したい、処理時間が読めない、プロキシ設定が要る

OSやPythonの更新を自分で行わないと止まる。Python 3.10のようにサポートが切れた版を放置するとセキュリティ修正が入らない

今回のように asyncio で1回数十秒〜数分に収まる処理なら、共有サーバーのcronでも動きます。一方、取得件数が増えて実行時間が延びる、失敗をログで追いたい、プロキシを環境変数で渡したいといった条件が揃うなら、systemd timerを使えるVPSか自宅の常時稼働機の方が扱いやすくなります。

並列化しても守ること：robots.txt・利用規約・アクセス間隔

コードに組み込んだもの

  • robots.txt：can_fetch() で拒否されたURLは取りに行かず、Crawl-delay があれば MIN_INTERVAL より長い方を採用しています。
  • アクセス間隔：並列化しても開始間隔は1秒以上。MAX_CONCURRENCY を上げても、相手から見た頻度は MIN_INTERVAL で決まります。
  • 相手の「待って」に従う：429・503は Retry-After を尊重し、リトライ回数にも上限を付けています。
  • 名乗る：User-Agentに連絡先を入れ、何者のアクセスかを相手が判断できるようにしています。

コードの外で確認すること

  • 利用規約：自動アクセスやデータ収集を禁じる条項がないか読む。日本ではスクレイピング自体が直ちに違法ではありませんが、規約違反はアカウント停止や契約上の問題になりえます。
  • 公式APIの有無：目的のデータにAPIがあるなら、そちらが最も安全で効率的です。
  • 個人情報：取得しないのが基本。必要なら利用目的の通知・公表が要ります。
  • ステータス200でも中身を確かめる：ボット対策のあるサイトでは、200で「Just a moment...」のような確認ページが返ることがあります。その場合はhttpxで取りにいく対象ではないと判断し、公式APIを探すか、相手に許可を求めます。

やらないこと

IPアドレスを次々に切り替える、ブラウザの指紋を偽装するなど、相手のアクセス制限や判定を回避する目的の書き方はしません。この記事でプロキシを扱ったのは、社内ネットワークの出口プロキシのように「自分の環境から外へ出るために必要な経路」としてだけです。社内CAで証明書エラーになる場合は verify=False で黙らせず、verify=False→truststore｜26年10月 の方法でOSの証明書ストアを使ってください。

関連する選択肢

毎日決まった時刻にスクリプトを動かすだけなら、共有のレンタルサーバーでも足ります。cronが使えるプランを選べば、自分でOSを管理する必要はありません。

レンタルサーバー エックスサーバー

※ 広告を含みます（A8.net）。リンク経由でお申し込みがあった場合、手数料を受け取ることがあります。

まとめ

httpx 0.28では proxies= が削除され、全通信を同じプロキシに通すなら proxy=、経路を分けるなら mounts= で書きます。mounts= のトランスポートはクライアントの limits= を引き継がないので、個別に渡すのを忘れないでください。並列取得は AsyncClient の Limits で接続数、asyncio.Semaphore で同時タスク数、自前の RateLimiter で開始間隔を、それぞれ別に絞ります。目的は応答待ちを重ねることで、相手へのリクエスト頻度は逐次版と変えません。gather(return_exceptions=True) で取れた分を保存しつつ失敗を数え、終了コードで返せば、systemd timer（Type=oneshot・Persistent=true・enable-linger）が毎日の成否を記録してくれます。Python 3.10は2026年10月1日でサポートが終わったので、venvを3.11以降で作り直すところから始めてください。