単にtweepyをラッピングしただけだけど、毎回、古いコードを漁って、使い方を思い出すのも面倒なんで、自分用Twitter APIライブラリと使い方のメモ。
My Twitter APIライブラリ
環境変数に以下の値を設定しとく。
-
TWITTER_CONSUMER_KEY
:Twitter Appの「Keys and Access Tokens」タブにある「Consumer Key (API Key)」 -
TWITTER_CONSUMER_SECRET
:Twitter Appの「Keys and Access Tokens」タブにある「Consumer Secret (API Secret)」 -
TWITTER_CALLBACK_URL
:Twitter Appの「Settings」タブで設定した「Callback URL」
twitter.py
# -*- coding: utf-8 -*-
import os
import urllib2
import os.path
import logging
import traceback
import oauthlib
import tweepy
CONSUMER_KEY = os.environ.get('TWITTER_CONSUMER_KEY')
CONSUMER_SECRET = os.environ.get('TWITTER_CONSUMER_SECRET')
CALLBACK_URL = os.environ.get('TWITTER_CALLBACK_URL')
#
def api(access_token = None, access_token_secret = None, with_callback_url = False):
if with_callback_url:
auth = tweepy.OAuthHandler(
consumer_key = CONSUMER_KEY,
consumer_secret = CONSUMER_SECRET,
callback_url = CALLBACK_URL)
else:
auth = tweepy.OAuthHandler(
consumer_key = CONSUMER_KEY,
consumer_secret = CONSUMER_SECRET)
if access_token is not None and access_token_secret is not None:
auth.set_access_token(access_token, access_token_secret)
return tweepy.API(auth)
# API
def get_tweets(screen_name = None, hashtag = None, count = 200):
if screen_name is not None:
cursor = tweepy.Cursor(api().user_timeline,
screen_name = screen_name, count = 200, exclude_replies = True)
elif hashtag is not None:
cursor = tweepy.Cursor(api().search, q = u"#{0}".format(hashtag), count = 200)
if count is not None:
return cursor.items(count)
else:
return cursor.items()
def tweet_link(tweet):
return "https://twitter.com/{screen_name}/status/{tweet_id}".format(
screen_name = tweet.user.screen_name,
tweet_id = tweet.id_str)
def get_friends(screen_name):
return tweepy.Cursor(api().friends, screen_name = screen_name, count = 200).items()
# OAuth
def generate_auth_url():
oauth_handler = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET, CALLBACK_URL)
redirect_url = oauth_handler.get_authorization_url(signin_with_twitter = True)
request_token = oauth_handler.request_token
return redirect_url, request_token['oauth_token'], request_token['oauth_token_secret'],
def get_access_token(request_token_key, request_token_secret, oauth_verifier):
oauth_handler = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
oauth_handler.request_token = {
'oauth_token': request_token_key,
'oauth_token_secret': request_token_secret,
'oauth_callback_confirmed': True,
}
oauth_handler.get_access_token(oauth_verifier)
return oauth_handler.access_token, oauth_handler.access_token_secret
def verify_credentials(access_token, access_token_secret):
api = api(access_token, access_token_secret)
return api.verify_credentials()
# Utility
def get_big_profile_image_url(profile_image_url):
return profile_image_url.replace('_normal', '')
WebでOAuth認証する手順
書いてるコードは、Bottle用のサンプルコード。
1. OAuth用のURLを発行する
@get('/twitter_login')
def twitter_login():
auth_url, request_token_key, request_token_secret = twitter.generate_auth_url()
# request_token_keyとrequest_token_secretをセッションに保存する。
redirect(auth_url)
twitter.generate_auth_url
を呼び出して、Twitterに飛ばすURLとリクエスト・トークン、シークレットを発行する。
そして、リクエスト・トークン、シークレットを、セッションに保存する。
ここで保存した値は、Twitterからのコールバックの処理でアクセストークンを取得する際に使う。
ここ、人によって実装が違うと思うんだけど、ぼくは、リクエスト・トークン、シークレットをデータベースに保存して、リクエスト・トークンだけを、クッキーに入れたりしてる。
あと、CSFR対策で、アクセス元のIPとかUser Agentも、データベースに入れることもある。
そして、Twitterにリダイレクト。
2. アクセストークンの取得
@get('/twitter_callback')
def twitter_callback():
# セッションに保存したリクエスト・トークン(request_token_key)、シークレット(request_token_secret)を取得する。
oauth_verifier = request.query['oauth_verifier']
access_token_key, access_token_secret = twitter.get_access_token(
twitter_connect.request_token_key,
twitter_connect.request_token_secret,
oauth_verifier)
twitter_user = twitter.verify_credentials(access_token_key, access_token_secret)
# ユーザー登録とかログインとかの処理
Twitterからのコールバックには、oauth_verifier
が入ってるんで、これとセッションに入れといたリクエスト・トークン、シークレットを使って、アクセス・トークン、シークレットを取得する。