Streaming

La X API admite datos en tiempo real a través de endpoints como el Filtered Stream Endpoint, que entrega los Posts coincidentes a medida que ocurren. Esto requiere realizar una conexión http persistente.

Configuración y streaming básico

Síncrono

from xdk import Client
# Initialize client
client = Client(bearer_token="your_bearer_token")
# Stream posts (make sure you have rules set up first)
for post_response in client.stream.posts():
    data = post_response.model_dump() if hasattr(post_response, 'model_dump') else dict(post_response)
    if 'data' in data and data['data']:
        tweet = data['data']
        post_text = tweet.get('text', '') if isinstance(tweet, dict) else (tweet.text if hasattr(tweet, 'text') else '')
        print(f"Post: {post_text}")

Asíncrono

Example

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())

Gestión de reglas

Las reglas definen filtros sobre los datos específicos que buscas (por ejemplo, palabras clave, usuarios, etc.). Puedes aprender más sobre cómo construir reglas en esta guía. Añadir reglas:

from xdk.stream.models import UpdateRulesRequest
# Add a rule
add_rules = {
    "add": [
        {"value": "from:xdevelopers", "tag": "official_updates"}
    ]
}
request_body = UpdateRulesRequest(**add_rules)
response = client.stream.update_rules(body=request_body)

Eliminar 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)

Listar reglas:

# get_rules returns an Iterator, so iterate over it
for page in client.stream.get_rules():
    if page.data:
        for rule in page.data:
            # Access rule attributes - Pydantic models support both attribute and dict access
            rule_id = rule.id if hasattr(rule, 'id') else rule.get('id', '')
            rule_value = rule.value if hasattr(rule, 'value') else rule.get('value', '')
            rule_tag = rule.tag if hasattr(rule, 'tag') else rule.get('tag', '')
            print(f"ID: {rule_id}, Value: {rule_value}, Tag: {rule_tag}")
    break  # Remove break to get all pages

Para la sintaxis completa de reglas, consulta X Streaming Rules Docs.

Solución de problemas

403 Forbidden: Autenticación inválida o permisos insuficientes.
420 Enhance Your Calm: Rate limited; espera y reintenta.
Sin datos: Verifica las reglas con get_rules(); asegúrate de que existan Posts que coincidan. Para ejemplos de código detallados usando el Python XDK, consulta nuestro repositorio de ejemplos de código en GitHub. Para más ejemplos y referencia de la API, consulta los docstrings en línea (por ejemplo, help(client.tweets.search_recent)) o los stubs generados en el código fuente. Contribuye con feedback a través del repositorio de GitHub.

Getting Started

Products

Fundamentals

Partners & Customers

Resources

Streaming

Configuración y streaming básico

Síncrono

Asíncrono

Gestión de reglas

Solución de problemas

​Configuración y streaming básico

​Síncrono

​Asíncrono

​Gestión de reglas

​Solución de problemas

Configuración y streaming básico

Síncrono

Asíncrono

Gestión de reglas

Solución de problemas