스트리밍

X API는 Filtered Stream Endpoint와 같은 엔드포인트를 통해 실시간 data를 지원하며, 발생 즉시 일치하는 Post를 전송합니다. 이를 위해 지속적인 HTTP 연결이 필요합니다.

​

설정 및 기본 스트리밍

​

동기식

from xdk import Client

# 클라이언트 초기화
client = Client(bearer_token="your_bearer_token")

# 포스트 스트리밍 (먼저 규칙을 설정해야 합니다)
for post_response in client.stream.posts():
    data = post_response.model_dump()
    
    if 'data' in data and data['data']:
        tweet = data['data']
        print(f"포스트: {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 참고 자료는 인라인 독스트링(예: help(client.tweets.search_recent)) 또는 소스에 생성된 스텁을 참조하세요. 피드백은 GitHub 저장소로 보내주세요.