More than 3 years have passed since last update.

Twitter API による連続取得 (の Tips)

Last updated at 2020-06-03Posted at 2020-06-01

Twitter API でツイートを取得する際、ご存知の通り、一定時間での取得回数や、一回の取得ツイート数が制限されています。

この制限とうまく付き合っていくために、連続取得する方法の Tips を記載しておきます。
なお、この投稿では、POST については扱いません。(ほぼ同じですが)

数値等は必ず一次情報をご確認ください。
サンプルコードは Gist にて公開しています。(Twitter パッケージがあるそうですが、使ってません)

Twitter API の Rate Limits (一定時間内の取得回数制限)

制限は、下のようになっています。

GET endpoints

The standard API rate limits described in this table refer to GET (read) endpoints. Note that endpoints not listed in the chart default to 15 requests per allotted user. All request windows are 15 minutes in length. These rate limits apply to the standard API endpoints only, does not apply to premium APIs.

Rate limits — Twitter Developers より抜粋

15分単位で取得回数に制限がかかります。
このページによれば、例えば "search/tweets" の制限は次の通り(あ、Standard です)。

Endpoint Resource family Requests / window (user auth) Requests / window (app auth)

GET search/tweets search 180 450

Endpoint	Resource family	Requests / window (user auth)	Requests / window (app auth)
GET search/tweets	search	180	450

Rate limits — Twitter Developers より抜粋

制限状況の取得

現在の制限の状況は https://api.twitter.com/1.1/application/rate_limit_status.json を Endpoint として取得できます。
パラメータの resources はオプションで、Resource family を指定します。

取得した情報例

user auth (OAuth v1.1)

{
  "rate_limit_context": {
    "access_token": "*******************"
  },
  "resources": {
    "search": {
      "/search/tweets": {
        "limit": 180,
        "remaining": 180,
        "reset": 1591016735
      }
    }
  }
}

app auth (OAuth v2.0)

{
  "rate_limit_context": {
    "application": "dummykey"
  },
  "resources": {
    "search": {
      "/search/tweets": {
        "limit": 450,
        "remaining": 450,
        "reset": 1591016736
      }
    }
  }
}

それぞれ、reset の数字が epoch time で、リセットされる時間が示されます。

上の例では、user auth (OAuth v1.1) で epoch time で 1591016735 = 2020-06-01 22:05:35、app auth (OAuth v2.0) で 1591016736 = 2020-06-01 22:05:36 にリセットされることを示しています。

制限に抵触すると、remaining の数値が 0 になります。

詳細は GET application/rate_limit_status — Twitter Developers に記載されています。("Example Response" は user auth の例のようです)

項目の整理

取得した情報の項目は次のようになっています。(search ファミリの例)

Category	Family	Endpoint	Key	Value
rate_limit_context			access_token (user auth (v1.1))	Access Token の内容
			application (app auth (v2.0))	dummykey (固定らしい)
resources
	search
		/search/tweets
			limit	制限時間内の最大回数
			remaining	制限時間内にアクセスできる残回数
			reset	制限時間がリセットされる時刻 (epoch time)

Resource Family を 'users' などの Endpoint が複数あるものにすると、複数の Endpoint の情報が一括で取得されます。
Resource Family を指定せずに取得すると、すべての Family が一括で取得されます。
誤った Resource Family を指定すると、resources がありません。

誤った Resource Family を指定したケース

下の例は Resource Family に誤って 'user' を指定した場合です。 (本当は 'users' (s あり) を指定すべきところ)

user auth

{
  "rate_limit_context": {
    "access_token": "*******************"
  }
}

app auth

{
  "rate_limit_context": {
    "application": "dummykey"
  }
}

いずれも "rate_limit_context" は返ってきますが、"resources" がありません。

Rate Limit エラー時のレスポンス(取得結果)

Rate Limit のエラー時には res.status_code (HTTP Status Code) に 429 が返ります。(420 が返ってくることもあります¹。)

Code Text Description

420 Enhance Your Calm Returned when an app is being rate limited for making too many requests.

429 Too Many Requests Returned when a request cannot be served due to the app's rate limit having been exhausted for the resource. See Rate Limiting.

Code	Text	Description
420	Enhance Your Calm	Returned when an app is being rate limited for making too many requests.
429	Too Many Requests	Returned when a request cannot be served due to the app's rate limit having been exhausted for the resource. See Rate Limiting.

Response codes — Twitter Developers より抜粋

【2020/06/03 追記】 420 については、下に詳しい説明がありました。

420 Rate Limited
The client has connected too frequently. For example, an endpoint returns this status if:

A client makes too many login attempts in a short period of time.

Too many copies of an application attempt to authenticate with the same credentials.

Connecting to a streaming endpoint — Twitter Developers より抜粋

JSON の errors.code に 88 が入ります。

{
  "errors": [
    {
      "code": 88,
      "message": "Rate limit exceeded"
    }
  ]
}

Code Text Description

88 Rate limit exceeded Corresponds with HTTP 429. The request limit for this resource has been reached for the current rate limit window.

Code	Text	Description
88	Rate limit exceeded	Corresponds with HTTP 429. The request limit for this resource has been reached for the current rate limit window.

Response codes — Twitter Developers より抜粋

requests などの例外については、各サイトをご覧ください。

処理の流れ

Rate Limit を考慮した処理の流れは次のような感じでしょうか。

while True:
    try:
        res = APIへのリクエストget/post
        res.raise_for_status()
    except requests.exceptions.HTTPError:
        # Rate Limit に到達すると 429/420 が返ってくる
        if res.status_code in (420, 429):
            Rate Limit 情報の取得               ← ここ
            reset 時刻までおとなしく待機
            continue
        420/429 以外の例外処理
    except OtherException:
        例外処理

    うまく取得できた時の処理
    break または return または yield など

エラーの処理は、GitHub にあるtwitterdev/twitter-python-ads-sdk: A Twitter supported and maintained Ads API SDK for Python. の図が参考になります。
図: Error Handling (敢えて表示させていません) ("429" が "Not Found" になっているのは、間違いかな、と思うのですが……)

以下は"Rate Limit 情報の取得" 部分の具体的な方策です。

制限情報の取得サンプル

GetTweetStatus クラス

情報取得のサンプルとしては、クラスで実装するメリットはあまりないのですが、実際にプログラムに組み込むことを考えると、モジュール化しやすい形で書いたほうが良さそうな気がするので、GetTweetStatus というクラスにしてあります。(apikey や Bearer など、できるだけ外からアクセスしないようにしておきたい、という気持ちもあり……)

class GetTweetStatus

    def __init__(self, apikey, apisec, access_token="", access_secret=""):
        self._apikey = apikey
        self._apisec = apisec
        self._access_token = access_token
        self._access_secret = access_secret
        self._access_token_mask = re.compile(r'(?P<access_token>"access_token":)\s*".*"')

最後の行の re.compile() は、受け取った access_token の表示をマスクするためのものです。

取得部分

user auth (OAuth v1.1)

GetTweetStatus.get_limit_status_v1( )

    def get_limit_status_v1(self, resource_family="search"):
        """OAuth v1.1 を使用したステータスの取得"""

        # OAuth は複雑なので OAuth1Session を利用する
        oauth1 = OAuth1Session(self._apikey, self._apisec, self._access_token, self._access_secret)

        params = {
            'resources': resource_family  # help, users, search, statuses etc.
        }

        try:
            res = oauth1.get(STATUS_ENDPOINT, params=params, timeout=5.0)
            res.raise_for_status()
        except (TimeoutError, requests.ConnectionError):
            raise requests.ConnectionError("Cannot get Limit Status")
        except Exception:
            raise Exception("Cannot get Limit Status")
        return res.json()

OAuth v1.1 は色々面倒なため、OAuth1Session() を利用しています。 requests_oauthlib のインストールと from requests_oauthlib import OAuth1Session が必要です。
例外処理は簡単に済ませています。

app auth (OAuth v2.0)

GetTweetStatus.get_limit_status_v2( )

    def get_limit_status_v2(self, resource_family="search"):
        """OAuth v2.0 (Bearer) を使用したステータスの取得"""
        bearer = self._get_bearer() # Bearer の取得

        headers = {
            'Authorization':'Bearer {}'.format(bearer),
            'User-Agent': USER_AGENT
        }
        params = {
            'resources': resource_family  # help, users, search, statuses etc.
        }

        try:
            res = requests.get(STATUS_ENDPOINT, headers=headers, params=params, timeout=5.0)
            res.raise_for_status()
        except (TimeoutError, requests.ConnectionError):
            raise requests.ConnectionError("Cannot get Limit Status")
        except Exception:
            raise Exception("Cannot get Limit Status")
        return res.json()

こちらは、requests を使用していますので、requests のインストールと import requests が必要です。
例外処理は簡単に済ませています。
1. 冒頭で Bearer Token を取得し、
2. これをヘッダに設定。
3. パラメータとして Resource Family を設定し、
4. STATUS_ENDPOINT へ送信しています。

Bearer の生成

bearer = self._get_bearer() # Bearer の取得 で呼び出している _get_bearer() 部分です。

GetTweetStatus._get_bearer( ), _get_credential( )

    def _get_bearer(self):
        """Bearer を取得"""
        cred = self._get_credential()
        headers = {
            'Authorization': 'Basic ' + cred,
            'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8',
            'User-Agent': USER_AGENT
            }
        data = {
            'grant_type': 'client_credentials',
            }

        try:
            res = requests.post(TOKEN_ENDPOINT, data=data, headers=headers, timeout=5.0)
            res.raise_for_status()
        except (TimeoutError, requests.ConnectionError):
            raise Exception("Cannot get Bearer")
        except requests.exceptions.HTTPError:
            if res.status_code == 403:
                raise requests.exceptions.HTTPError("Auth Error")
            raise requests.exceptions.HTTPError("Other Exception")
        except Exception:
            raise Exception("Cannot get Bearer")
        rjson = res.json()
        return rjson['access_token']

    def _get_credential(self):
        """Credential の生成"""
        pair = self._apikey + ':' + self._apisec
        bcred = b64encode(pair.encode('utf-8'))
        return bcred.decode()

"Application-only authentication — Twitter Developers" に書かれている通り設定しています。
1. APIKEY と APISEC を組み合わせて Base64 エンコードを行い、
2. ヘッダーに設定し、
3. ペイロード data に grant_type="client_credentials" を設定して
4. Endpoint へリクエスト(POST)しています。
5. 戻ってきた JSON の "access_token" に Bearer Token が設定されていますので、これを取得します。

表示部分

メソッドとして実装しています。
実際の利用時には、 "reset" を返すようなものに実装するところでしょうか。

GetTweetStatus.disp_limit_status( )

    def disp_limit_status(self, version=2, resource_family="search"):
        """バージョンに分けて Rate Limit を表示する"""
        if version == 2:
            resj = self.get_limit_status_v2(resource_family=resource_family)
        elif version == 1:
            resj = self.get_limit_status_v1(resource_family=resource_family)
        else:
            raise Exception("Version error: {version}")

        # JSON の表示
        print(self._access_token_mask.sub(r'\g<access_token> "*******************"',
                                          json.dumps(resj, indent=2, ensure_ascii=False)))
        # 分解表示(remain/reset の取得例)
        print("resources:")
        if 'resources' in resj:
            resources = resj['resources']
            for family in resources:
                print(f"  family: {family}")
                endpoints = resources[family]
                for endpoint in endpoints:
                    items = endpoints[endpoint]
                    print(f"    endpoint: {endpoint}")
                    limit = items['limit']
                    remaining = items['remaining']
                    reset = items['reset']
                    e2d = epoch2datetime(reset)
                    duration = get_delta(reset)
                    print(f"      limit: {limit}")
                    print(f"      remaining: {remaining}")
                    print(f"      reset: {reset}")         # ← 実際にはこれを返すような形
                    print(f"      reset(epoch2datetime): {e2d}")
                    print(f"      duration: {duration} sec")
        else:
            print("  Not Available")

前半で、指定したプロトコル(user auth/app auth)で情報を取得し、
後半で分解して表示しています。
実際に使用する際には、remaining や reset を用いて、対応することになります。

時刻関連のユーティリティ & ヘッダ部

時刻操作のユーティリティとファイルの冒頭部分です。

getTwitterStatus.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Twitter Rate Limit 情報取得サンプル"""

import os
import sys
import json
from base64 import b64encode
import datetime
import time
import re
import argparse

#!pip install requests
import requests
#!pip install requests_oauthlib
from requests_oauthlib import OAuth1Session


USER_AGENT = "Get Twitter Staus Application/1.0"
TOKEN_ENDPOINT = 'https://api.twitter.com/oauth2/token'
STATUS_ENDPOINT = 'https://api.twitter.com/1.1/application/rate_limit_status.json'


def epoch2datetime(epoch):
    """エポックタイム (UNIX タイム) を datetime (localtime) へ変換"""
    return datetime.datetime(*(time.localtime(epoch)[:6]))


def datetime2epoch(d_utc):
    """datetime (UTC) をエポックタイム (UNIX タイム)へ変換"""
    #UTC を localtime へ変換
    date_localtime = \
        d_utc.replace(tzinfo=datetime.tzinfo.tz.tzutc()).astimezone(datetime.tzinfo.tz.tzlocal())
    return int(time.mktime(date_localtime.timetuple()))


def get_delta(target_epoch_time):
    """target_epoch_time と現在時刻の差分を返す"""
    return target_epoch_time - int(round(time.time(), 0))

main() 部分

せっかくなので、コマンドラインの引数で OAuth バージョンや Resource Family を指定できるようにしてみました。

main( )

def main():
    """main()"""
    # API_KEY, API_SEC 等環境変数の確認
    apikey = os.getenv('API_KEY', default="")
    apisec = os.getenv('API_SEC', default="")
    access_token = os.getenv('ACCESS_TOKEN', default="")
    access_secret = os.getenv('ACCESS_SECRET', default="")

    if apikey == "" or apisec == "":    # 環境変数が取得できない場合
        print("環境変数 API_KEY と API_SEC を設定してください。", file=sys.stderr)
        print("OAuth v1.1 を使用する場合には環境変数 ACCESS_TOKEN と ACCESS_SECRET も設定してください。",
              file=sys.stderr)
        sys.exit(255)

    # 引数の設定
    parser = argparse.ArgumentParser()
    parser.add_argument('-a', '--oauthversion', type=int, default=0,
                        metavar='N', choices=(0, 1, 2),
                        help=u'OAuth のバージョン指定 [1|2]')
    parser.add_argument('-f', '--family', type=str, default='search',
                        metavar='Family',
                        help=u'API ファミリの指定。複数の場合はカンマ区切り')

    args = parser.parse_args()
    oauthversion = args.oauthversion
    family = args.family

    # GetTweetStatus オブジェクトの取得
    gts = GetTweetStatus(apikey, apisec, access_token=access_token, access_secret=access_secret)

    # User Auth (OAuth v1.1) による Rate Limit 取得と表示
    if (oauthversion in (0, 1)) and (access_token != "" and access_secret != ""):
        print("<<user auth (OAuth v1)>>")
        gts.disp_limit_status(version=1, resource_family=family)

    # App Auth (OAuth v2.0) による Rate Limit 取得と表示
    if oauthversion in (0, 2):
        print("<<app auth (OAuth v2)>>")
        gts.disp_limit_status(version=2, resource_family=family)

if __name__ == "__main__":
    main()

API key などは、引数に渡したくはないので、環境変数から読み込むようにしてあります。
API_KEY: Consumer API key (必須)
API_SEC: Consumer API secret key (必須)
ACCESS_TOKEN: Access token (v1 利用時には必要)
ACCESS_SECRET: Access token secret (同上)

全コード `getTwitterStatus.py`

コード全体は、Gist に置いてみました²。
内包表記などが苦手なので、冗長かもしれませんが、改善点をコメントなどいただけましたら幸いです。
Twitter 制限情報の取得サンプル - Twitter API による連続取得 (の Tips) - Gist

実行結果

$ python3 getTwitterStatus.py
<<user auth (OAuth v1)>>
{
  "rate_limit_context": {
    "access_token": "*******************"
  },
  "resources": {
    "search": {
      "/search/tweets": {
        "limit": 180,
        "remaining": 180,
        "reset": 1591016735
      }
    }
  }
}
resources:
  family: search
    endpoint: /search/tweets
      limit: 180
      remaining: 180
      reset: 1591016735
      reset(epoch2datetime): 2020-06-01 22:05:35
      duration: 899 sec
<<app auth (OAuth v2)>>
{
  "rate_limit_context": {
    "application": "dummykey"
  },
  "resources": {
    "search": {
      "/search/tweets": {
        "limit": 450,
        "remaining": 450,
        "reset": 1591016736
      }
    }
  }
}
resources:
  family: search
    endpoint: /search/tweets
      limit: 450
      remaining: 450
      reset: 1591016736
      reset(epoch2datetime): 2020-06-01 22:05:36
      duration: 900 sec
$

取得ツイート数の制限

時間内の制限とは関係なく、一回に取得できる回数にも制限があります。

statuses/user_timeline の場合は 200 ($count\leq200$)、search/tweets の場合は 100 ($count\leq100$)。
他にも色々と制限がありますが、search/tweets の場合には、続けて取得が可能なように、search_metadata 内に next_results という項目が入ってきます。

search/tweets の場合

Standard search API — Twitter Developers の Example Response より抜粋

{
  "statuses": [
  ...
  ],
  "search_metadata": {
    "completed_in": 0.047,
    "max_id": 1125490788736032770,
    "max_id_str": "1125490788736032770",
    "next_results": "?max_id=1124690280777699327&q=from%3Atwitterdev&count=2&include_entities=1&result_type=mixed",
    "query": "from%3Atwitterdev",
    "refresh_url": "?since_id=1125490788736032770&q=from%3Atwitterdev&result_type=mixed&include_entities=1",
    "count": 2,
    "since_id": 0,
    "since_id_str": "0"
  }
}

search_metadata 内に next_results がありますので、これを新しいパラメータとしてリクエストすると、残りのサーチ結果も取得できます(count に指定した単位で)。

時間内の回数制限にかからない間は、これを参照して繰り返すと、結果を連続して取得できます。
つまり、$count$(最大100)$×limit$(user auth の場合は180)$＝18,000 ツイート$ を Rate Limit の中で取得できます。

上のサンプルの場合、$count=2$ となっていますので、このまま続けると $count(2)ツイート/回 × limit(180)回/15分=360ツイート/15分$ の取得で、制限に到達します(もちろん、がんがんリクエストした場合です)。

search の結果がすべて取得し終わると、search_metadata の中から next_results が消えます。

なお、時々、再取得させると next_results が復活することがあるので、少し待って、リトライするといいかもしれません。

metadata がない場合

statuses/user_timeline などの場合には、*_metadata は含まれていませんので、max_id の指定をうまく利用して、search の next_results に相当するようなものを自分で生成する必要があります。(実は、search 以外に利用したことがないので、よくわかっていませんが、そんなに外れていはいないのではないかと)

search の場合には過去7日間が対象となっていますが、user_timeline は過去24時間なので、そもそも目的が異なるとは思いますが……

まとめ

Twitter API を使用する際の制限について、レスポンスの状態と制限情報(Rate Limit)の取得について書きました。
Rate Limit Status のみを取得するサンプルコードを公開しました。
search 時の metadata と利用方法について記載しました。
search 以外は、ほとんど利用したことがないので資料ベースで書きました。
公開したコードについては、積極的にコメントいただければ幸いです。

参考 (本稿で参照したものをベースに)

一般に 429 の "Too Many Requests: Returned when a request cannot be served due to the app's rate limit having been exhausted for the resource." が返るのですが、極まれに 420 の "Enhance Your Calm: Returned when an app is being rate limited for making too many requests." が返って来ることがあります。後者は誤って複数のリクエストを同時に投げてしまった場合に起きるのかもしれません(未検証)。→ "Connecting to a streaming endpoint — Twitter Developers" に説明がありましたので、本文に追記しました(2020/06/03)。 ↩
初めて Gist を使ってみました。使い方が合っているのか、不安です。 ↩

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

Twitter API による連続取得 (の Tips)

Twitter API の Rate Limits (一定時間内の取得回数制限)

GET endpoints

制限状況の取得

取得した情報例

項目の整理

誤った Resource Family を指定したケース

Rate Limit エラー時のレスポンス(取得結果)

処理の流れ

制限情報の取得サンプル

GetTweetStatus クラス

取得部分

user auth (OAuth v1.1)

app auth (OAuth v2.0)

Bearer の生成

表示部分

時刻関連のユーティリティ & ヘッダ部

main() 部分

全コード getTwitterStatus.py

実行結果

取得ツイート数の制限

search/tweets の場合

metadata がない場合

まとめ

参考 (本稿で参照したものをベースに)

全コード `getTwitterStatus.py`