X APIを使って投稿する
使用技術
ライブラリ
tweepy (Python Twitter API wrapper)
requestやresponseはFastAPIのやつ
認証方式: OAuth 1.0a
画像アップロードのためにOAuth 1.0aを使用
OAuth 2.0ではMedia Upload APIが利用不可
API Key
Consumer Keysを使用する
認証フロー
code:python
from authlib.integrations.starlette_client import OAuth
# Twitterクライアント設定
oauth = OAuth()
oauth.register(
name="twitter",
client_id=os.getenv("TWITTER_CONSUMER_ID"),
client_secret=os.getenv("TWITTER_CONSUMER_SECRET"),
)
# 認証開始
redirect_uri = request.url_for("auth_callback", provider="twitter")
return await oauth.twitter.authorize_redirect(request, redirect_uri)
# コールバック処理
token = await oauth.twitter.authorize_access_token(request)
resp = await oauth.twitter.get("account/verify_credentials.json", token=token)
user_data = resp.json()
# セッションに保存
account_info = {
"id": user_data.get("id_str"),
"username": user_data.get("screen_name"),
"name": user_data.get("name"),
"token": token, # oauth_token と oauth_token_secret を含む
}
投稿処理
code:python
import tweepy
async def post_to_twitter(
text: str,
"""
OAuth 1.0aで画像付きツイートを投稿
Args:
token: OAuth 1.0a トークン (oauth_token, oauth_token_secret)
text: ツイート本文
"""
# 認証情報取得
consumer_key = os.getenv("TWITTER_CONSUMER_ID")
consumer_secret = os.getenv("TWITTER_CONSUMER_SECRET")
access_token = token.get("oauth_token")
access_token_secret = token.get("oauth_token_secret")
# ステップ1: 画像アップロード (API v1.1を使用)
auth = tweepy.OAuth1UserHandler(
consumer_key,
consumer_secret,
access_token,
access_token_secret
)
api = tweepy.API(auth)
media_ids = []
if images:
for image_bytes, mime_type in images:
file_obj = io.BytesIO(image_bytes)
filename = _get_filename_from_mime_type(mime_type)
media = api.media_upload(filename=filename, file=file_obj)
media_ids.append(media.media_id_string)
# ステップ2: ツイート投稿 (API v2を使用)
client = tweepy.Client(
consumer_key=consumer_key,
consumer_secret=consumer_secret,
access_token=access_token,
access_token_secret=access_token_secret,
)
resp = client.create_tweet(
text=text,
media_ids=media_ids if media_ids else None
)
return resp.data
def _get_filename_from_mime_type(mime_type: str) -> str:
"""Get appropriate filename based on MIME type."""
if "png" in mime_type:
return "image.png"
elif "gif" in mime_type:
return "image.gif"
return "image.jpg"
レート制限監視
code:python
from datetime import datetime
import logging
from typing import Any
logger = logging.getLogger(__name__)
def _log_rate_limit_info(response: Any, endpoint: str) -> None:
"""APIレスポンスヘッダからレート制限情報を取得"""
headers = response._headers or response.headers
limit = headers.get("x-rate-limit-limit")
remaining = headers.get("x-rate-limit-remaining")
reset = headers.get("x-rate-limit-reset")
if reset:
reset_dt = datetime.fromtimestamp(int(reset))
logger.info(
f"{remaining}/{limit} requests remaining, "
f"resets at {reset_dt}"
)