ストリーミング

X API は、Filtered Stream Endpoint などのエンドポイントを通じてリアルタイムデータを提供し、条件に一致する Post を発生と同時に配信します。これには持続的な HTTP 接続が必要です。

​

セットアップと基本的なストリーミング

​

同期処理

from xdk import Client

# クライアントを初期化する
client = Client(bearer_token="your_bearer_token")

# Post をストリーミング（事前にルールを設定しておいてください）
for post_response in client.stream.posts():
    data = post_response.model_dump()
    
    if 'data' in data and data['data']:
        tweet = data['data']
        print(f"Post: {tweet.get('text', '')}")

​

非同期処理

import asyncio
from asyncio import Queue
import threading
from xdk import Client

async def stream_posts_async(client: Client):
    queue = Queue()
    loop = asyncio.get_event_loop()
    stop = threading.Event()
    
    def run_stream():
        for post in client.stream.posts():
            if stop.is_set():
                break
            asyncio.run_coroutine_threadsafe(queue.put(post), loop)
        asyncio.run_coroutine_threadsafe(queue.put(None), loop)
    
    threading.Thread(target=run_stream, daemon=True).start()
    
    while True:
        post = await queue.get()
        if post is None:
            break
        data = post.model_dump()
        if 'data' in data and data['data']:
            print(f"Post: {data['data'].get('text', '')}")
    stop.set()

async def main():
    client = Client(bearer_token="your_bearer_token")
    await stream_posts_async(client)

asyncio.run(main())

​

ルール管理

ルールは、取得対象となる特定のdataに適用するフィルターを定義します（例：キーワード、ユーザーなど）。ルールの作成方法についてはこちらのガイドをご覧ください。

ルールの追加:

from xdk.stream.models import UpdateRulesRequest

# ルールを追加する
add_rules = {
    "add": [
        {"value": "from:xdevelopers", "tag": "公式アップデート"}
    ]
}

request_body = UpdateRulesRequest(**add_rules)
response = client.stream.update_rules(body=request_body)

ルールの削除:

from xdk.stream.models import UpdateRulesRequest

delete_rules = {
    "delete": {
        "ids": ["rule_id_1", "rule_id_2"]
    }
}

request_body = UpdateRulesRequest(**delete_rules)
response = client.stream.update_rules(body=request_body)

掲載規定:

response = client.stream.get_rules()

# ルールを表示
for rule in response.data:
    print(f"ID: {rule.id}, Value: {rule.value}, Tag: {rule.tag}")

ルール構文の詳細は、X Streaming Rules ドキュメントをご参照ください。

​

トラブルシューティング

• 403 Forbidden: 認証が無効、または権限が不足しています。
• 420 Enhance Your Calm: レート制限中です。しばらく待ってから再試行してください。
• No Data: get_rules() でルールを確認し、一致する Post が存在することを確認してください。

より多くの例や API リファレンスは、インラインの docstring（例: help(client.tweets.search_recent)) またはソース内の生成スタブを参照してください。フィードバックは GitHub リポジトリ までお寄せください。