Streaming

La X API admite datos en tiempo real mediante endpoints como el Filtered Stream Endpoint, que entregan Posts que coinciden a medida que se publican. Esto requiere establecer una conexión HTTP persistente.

​

Configuración y transmisión básicas

​

Sincronizado

from xdk import Client

# Inicializar cliente
client = Client(bearer_token="your_bearer_token")

# Transmitir Posts en streaming (asegúrate de tener reglas configuradas primero)
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', '')}")

​

Asíncrono

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="tu_bearer_token")
    await stream_posts_async(client)

asyncio.run(main())

​

Gestión de reglas

Las reglas definen filtros para los datos específicos que busca (p. ej., palabras clave, usuarios, etc.). Puede obtener más información sobre cómo crear reglas en esta guía

Agregar reglas:

from xdk.stream.models import UpdateRulesRequest

# Añadir una regla
add_rules = {
    "add": [
        {"value": "from:xdevelopers", "tag": "actualizaciones_oficiales"}
    ]
}

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

Eliminación de reglas:

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)

Reglas de publicación:

response = client.stream.get_rules()

# Imprimir reglas
for rule in response.data:
    print(f"ID: {rule.id}, Valor: {rule.value}, Etiqueta: {rule.tag}")

Para ver la sintaxis completa de las reglas, consulta la documentación de reglas de streaming de X.

​

Solución de problemas

• 403 Forbidden: Autenticación no válida o permisos insuficientes.
• 420 Enhance Your Calm: Has alcanzado el límite de frecuencia; espera y vuelve a intentarlo.
• No Data: Revisa las reglas con get_rules(); asegúrate de que existan Posts coincidentes.

Para más ejemplos y la referencia de la API, consulta las docstrings integradas (p. ej., help(client.tweets.search_recent)) o los stubs generados en el código fuente. Envía tus comentarios a través del repositorio de GitHub.