Pythonでrequestsライブラリを使ってURLを短縮する方法

PythonでURLを短縮するのに必要なのは、HTTP POST一回だけです。requestsライブラリを使って長いリンクを短縮サービスのAPIに送信し、APIキーをBearerトークンとして渡し、JSONレスポンスから短縮リンクを読み取ります。全体はおよそ5行に収まりますが、本ガイドの残りの部分では、実運用に乗せる際に加えるべきもの、すなわちエラーハンドリング、冪等性(idempotency)、一括処理、非同期処理を扱います。

本稿はURLを短縮する方法についての一般的な解説の開発者向けバージョンです。あちらの記事はダッシュボードとブラウザでの操作を扱っていますが、こちらはスクリプトにそのまま貼り付けられるコードを扱います。以下のエンドポイントとフィールド名はElidoのAPIのものですが、遷移先URLをPOSTして短縮URLを受け取るという構造自体は、ほとんどのモダンな短縮サービスで共通しているため、このパターンはそのまま応用できます。

まだサービスを選んでいない場合は、無料で使えるURL短縮APIの概要から始めてください。本稿が前提とするリクエストの形式、認証モデル、無料枠の上限がそこで説明されています。

最速の方法: requestsによるPOST一回で完結させる

requestsをインストールし、APIキーを環境変数に入れ、遷移先URLをlinksエンドポイントにPOSTします。

import os
import requests

resp = requests.post(
    "https://api.elido.app/v1/links",
    headers={"Authorization": f"Bearer {os.environ['ELIDO_API_KEY']}"},
    json={"destination_url": "https://example.com/a-very-long-path?with=params"},
    timeout=10,
)
resp.raise_for_status()
print(resp.json()["short_url"])   # -> https://s.elido.me/ab12cd

これをおもちゃのコードではなく本番運用に耐えるものにしている要素が3つあります。APIキーは環境変数から取得しており、ソースコードに直書きされていません。timeoutが設定されているため、接続がハングしても永遠にブロックされるのではなく、素早く失敗します。そしてraise_for_status()が4xxや5xxを、キャッチ可能な例外に変換してくれるため、失敗した呼び出しが成功したように見えてしまうことがありません。

レスポンスはJSONです。完成したリンクはshort_urlから読み取ります。多くの短縮サービスは、後でリンクを編集したり失効させたりする予定がある場合に保存しておくidも併せて返します。

エラーとレート制限を適切に処理する

うまくいくケースだけなら5行で済みます。しかし人手を介さず動くスクリプトは、うまくいかないケース、すなわち一時的な5xx、リクエストが速すぎるときの429、ネットワークの瞬断にも耐える必要があります。リトライする価値のあるエラーではリトライし、絶対に成功しないエラーでは諦めるように、呼び出しをラップしましょう。

import os
import time
import requests

def shorten(destination: str, *, retries: int = 3) -> str:
    headers = {"Authorization": f"Bearer {os.environ['ELIDO_API_KEY']}"}
    for attempt in range(retries):
        resp = requests.post(
            "https://api.elido.app/v1/links",
            headers=headers,
            json={"destination_url": destination},
            timeout=10,
        )
        if resp.status_code == 429:
            time.sleep(int(resp.headers.get("Retry-After", 2)))
            continue
        if resp.status_code >= 500:
            time.sleep(2 ** attempt)   # exponential backoff
            continue
        resp.raise_for_status()        # 4xx other than 429 -> raise
        return resp.json()["short_url"]
    raise RuntimeError(f"shorten failed after {retries} attempts")

401や403は、リトライしても自然に解消することは決してないため、raise_for_status()まで素通りしてスクリプトを停止させます。429の場合はAPIが返すRetry-Afterヘッダーに従います。5xxの場合は指数関数的にバックオフします。何も考えずにリトライすることが、なぜ1回の障害を2回の障害に変えてしまうのかについては、レート制限と冪等性の詳細解説で扱っています。

リトライで重複が起きないようIdempotency-Keyを送信する

先ほどのリトライループには、見落としがちなバグがあります。POSTがサーバー側では成功しても、レスポンスが返ってくる途中で失われた場合、コード側はタイムアウトと判断してリトライし、同じURLに対して二つ目の短縮リンクを作ってしまいます。Idempotency-Keyはこれを解決します。論理的に一つのリクエストごとに安定した一意のキーを送ることで、APIは繰り返しの呼び出しに対して新しいリンクを発行するのではなく、元のリンクを返してくれます。

import uuid

key = str(uuid.uuid4())   # one key per URL you want shortened once
resp = requests.post(
    "https://api.elido.app/v1/links",
    headers={
        "Authorization": f"Bearer {os.environ['ELIDO_API_KEY']}",
        "Idempotency-Key": key,
    },
    json={"destination_url": "https://example.com/launch"},
    timeout=10,
)

キーはURLごとに一度だけ生成し、同じURLのリトライ全体で使い回してください。試行のたびに生成してはいけません。バッチ自体を再実行する可能性があるなら、URLと一緒にキーも保存しておきましょう。これは、1件のリンクを短縮する段階からリストを短縮する段階へ移るときに最も重要な習慣であり、まさにこの理由からAPIとSDKにも組み込まれています。

URLを一括で短縮する

安全なshorten()関数さえあれば、バッチ処理はただのループです。ただし、レート制限を尊重し、入力と出力の対応関係を失わないループである必要があります。

urls = [
    "https://example.com/spring-sale",
    "https://example.com/newsletter",
    "https://example.com/docs/getting-started",
]

results = {}
for url in urls:
    try:
        results[url] = shorten(url)
    except Exception as err:      # log and keep going; one bad URL should not sink the batch
        results[url] = f"ERROR: {err}"

for original, short in results.items():
    print(f"{short}\t{original}")

結果を元のURLをキーとした辞書に保持しておけば、一部が失敗しても、それが見える形で残り、再実行も可能になります。サイレントな欠落にはなりません。数百件程度のリンクであれば、この逐次処理版で十分です。数千件になると往復のレイテンシが積み重なっていき、そこで非同期処理が真価を発揮します。

httpxによる大規模な非同期処理

バッチが大きくなると、ボトルネックになるのはPython自体ではなく、ネットワークの待ち時間です。httpxのような非同期クライアントを使えば、同時に処理中のリクエスト数に上限を設けながら多数のリクエストを並行送信できるため、レート制限に引っかかることなく接続を目一杯活用できます。

import asyncio
import os
import httpx

async def shorten_all(urls: list[str], concurrency: int = 10) -> dict[str, str]:
    headers = {"Authorization": f"Bearer {os.environ['ELIDO_API_KEY']}"}
    limit = asyncio.Semaphore(concurrency)
    out: dict[str, str] = {}

    async with httpx.AsyncClient(timeout=10) as client:
        async def one(url: str) -> None:
            async with limit:
                r = await client.post(
                    "https://api.elido.app/v1/links",
                    headers=headers,
                    json={"destination_url": url},
                )
                out[url] = r.json()["short_url"]

        await asyncio.gather(*(one(u) for u in urls))
    return out

# asyncio.run(shorten_all(my_urls))

Semaphore(10)こそがこのコードの要です。同時実行リクエスト数を10件に制限することで、APIを429の連発に追い込むことなく高速に処理できます。この値は契約プランに明記された上限に合わせて調整してください。リクエストの正確な形式、フィールド名、現在の上限についてはAPIドキュメントが正式なリファレンスであり、クライアントを自前で実装したくない場合は開発者向けソリューションにSDKが用意されています。

どのアプローチを選ぶべきか

ツールは作業の規模に合わせて選びましょう。スクリプト内でリンクを1件だけ短縮するなら、5行のrequests呼び出しで十分です。スケジュール実行される信頼性の高いジョブには、リトライとIdempotencyを組み合わせたバージョンを使います。数百件程度の単発バッチには逐次ループが向いています。期限内に数万件をこなす必要があるなら、semaphoreを備えたhttpxの出番です。

どれを選ぶにしても、キーは環境変数に保持し、タイムアウトを設定し、リトライされる可能性のある処理にはIdempotency-Keyを送りましょう。この三つの習慣こそが、デモ用のスニペットと、放っておいても動き続けるスクリプトとの違いを生みます。

コーナーストーン記事シリーズを読む

本稿はengineeringクラスターに属しています。出発点となるのは、エンドポイントの形式と認証について解説する無料で使えるURL短縮APIガイドで、その次に負荷がかかる状況で適切に振る舞うためのレート制限と冪等性の記事があります。実際に動くリファレンスはAPIドキュメントです。

関連記事

• 無料で使えるURL短縮API: リクエストの形式・認証・上限
• URL短縮APIのレート制限と冪等性
• URLを短縮する方法: 無料・ブランド・一括のリンク