Prefect

Crea flujos de trabajo programados de búsqueda de Twitter, perfiles, líneas de tiempo y tendencias en Python con tareas de solo lectura de Prefect y TwexAPI.

Prefect es un orquestador de flujos de trabajo de código abierto para Python. Usa prefect-x-api-scraper para búsquedas de tweets repetibles, consultas de perfiles, actualizaciones de líneas de tiempo y verificaciones de tendencias.

La colección es de solo lectura. Proporciona seis tareas asíncronas de Prefect. Cada tarea devuelve la respuesta JSON canónica de TwexAPI como un diccionario de Python.

Usa esta colección para investigación, enriquecimiento, dashboards, alertas e indexación. Usa REST directo, el Python SDK o MCP para escrituras, paginación de seguidores y exportaciones.

Instalación

Usa Python 3.10 o posterior.

python -m pip install "prefect>=3.0.0" "prefect-x-api-scraper"

Registra el bloque de credenciales después de instalar Prefect:

prefect block register -m prefect_x_api_scraper

Guarda la API key de TwexAPI

Crea una API key desde el panel de TwexAPI. Guárdala dentro de un bloque TwexApiCredentials.

from prefect_x_api_scraper import TwexApiCredentials

credentials = TwexApiCredentials(
    api_key="YOUR_API_KEY",
    base_url="https://api.twexapi.io",
    timeout_seconds=30,
)
credentials.save("twexapi", overwrite=True)

Los bloques de Prefect almacenan configuración tipada entre flujos y despliegues. Nunca coloques API keys en YAML de despliegue, parámetros de flujo, logs o repositorios.

Elige la tarea de Prefect correcta

Construye un flujo de automatización de Twitter en Python

Este flujo busca publicaciones recientes y normaliza filas de tweets. Conserva IDs, autores, marcas de tiempo, métricas, URLs y estado de paginación.

from __future__ import annotations

from typing import Any

from prefect import flow, task
from prefect_x_api_scraper import TwexApiCredentials, search_tweets


@task
async def normalize_tweet_page(page: dict[str, Any]) -> list[dict[str, Any]]:
    rows: list[dict[str, Any]] = []
    for tweet in page.get("data", {}).get("tweets", page.get("tweets", [])):
        if not isinstance(tweet, dict):
            continue
        author = tweet.get("author") or tweet.get("user")
        rows.append(
            {
                "tweet_id": tweet.get("id") or tweet.get("tweet_id"),
                "text": tweet.get("text") or tweet.get("full_text"),
                "created_at": tweet.get("created_at") or tweet.get("createdAt"),
                "author": author if isinstance(author, dict) else {},
                "like_count": tweet.get("like_count") or tweet.get("likeCount"),
                "repost_count": tweet.get("retweet_count") or tweet.get("retweetCount"),
                "reply_count": tweet.get("reply_count") or tweet.get("replyCount"),
            }
        )
    return rows


@flow(name="TwexAPI Twitter Search")
async def social_signal_flow() -> dict[str, Any]:
    credentials = TwexApiCredentials.load("twexapi")
    page = await search_tweets(
        credentials,
        '"workflow orchestration" lang:en -filter:retweets',
        sort_by="Latest",
    )
    rows = await normalize_tweet_page(page)
    return {
        "tweet_rows": rows,
        "has_more": page.get("has_more") or page.get("has_next_page", False),
        "next_cursor": page.get("next_cursor") or page.get("nextCursor"),
    }

Ejecútalo con asyncio:

import asyncio

result = asyncio.run(social_signal_flow())

Escribe consultas de búsqueda de tweets enfocadas

Usa sort_by="Latest" para monitoreo cronológico. Usa sort_by="Top" para descubrimiento ordenado por engagement. Persiste los IDs de tweet porque los rankings pueden cambiar.

Programa el pipeline de búsqueda de Twitter

Usa .serve() para un proceso local de larga duración.

from prefect.schedules import Cron


if __name__ == "__main__":
    social_signal_flow.serve(
        name="twexapi-social-signals",
        schedule=Cron("0 * * * *", timezone="UTC"),
    )

Usa un despliegue con work pool para workers de Docker, Kubernetes o serverless.

prefect deploy social_signal_flow.py:social_signal_flow \
  --name twexapi-social-signals \
  --pool production

Pagina resultados de tweets y perfiles

La paginación por cursor continúa búsquedas grandes sin adivinar números de página. Mantén la solicitud original sin cambios. Pasa solo el cursor devuelto.

from typing import Any, Optional

from prefect import flow
from prefect_x_api_scraper import TwexApiCredentials, search_tweets


@flow
async def collect_tweet_pages(query: str) -> list[dict[str, Any]]:
    credentials = TwexApiCredentials.load("twexapi")
    rows_by_id: dict[str, dict[str, Any]] = {}
    cursor: Optional[str] = None

    while True:
        page = await search_tweets(
            credentials,
            query=query,
            sort_by="Latest",
            cursor=cursor,
        )

        tweets = page.get("data", {}).get("tweets", page.get("tweets", []))
        for tweet in tweets:
            if isinstance(tweet, dict):
                tweet_id = tweet.get("id") or tweet.get("tweet_id")
                if tweet_id:
                    rows_by_id[str(tweet_id)] = tweet

        cursor_value = page.get("next_cursor") or page.get("nextCursor")
        has_more = bool(page.get("has_more") or page.get("has_next_page", False))
        if not has_more or not cursor_value:
            break

        cursor = str(cursor_value)

    return list(rows_by_id.values())

No decodifiques cursores. Trátalos como cadenas opacas. Guarda cada cursor después de persistir su página.

Haz que las ejecuciones programadas sean idempotentes

Reintenta solo fallos transitorios

Prefect soporta retrasos de reintento, jitter y reintentos condicionales. No reintentes entradas inválidas sin cambios.

from prefect_x_api_scraper import search_tweets

search_recent_tweets = search_tweets.with_options(
    name="Search Recent Tweets",
    retries=4,
    retry_delay_seconds=[5, 15, 45, 120],
)

Enruta errores documentados

Consulta Error Handling y Rate Limits para orientación completa de recuperación. | 5xx | Reintenta fallos transitorios con un límite. |

Alternativa de planificación con MCP

Usa Twexapi MCP durante la planificación para descubrir endpoints, luego codifica la ruta seleccionada en tareas de Prefect o llamadas REST.

Prompt de planificación sugerido:

Use Twexapi MCP explore to choose the best endpoint for daily AI trend collection.
Return the exact method, relative path, query parameters, pagination fields, expected response fields, and retry guidance.
Do not execute write endpoints.

Colección de Prefect o API directa de TwexAPI

Elige prefect-x-api-scraper por sus seis lecturas compatibles. Proporciona bloques, llamadas asíncronas, validación y metadatos de tareas.

Usa REST directo, el Python SDK o MCP para:

• Acciones de tweet, respuesta, cita, like, follow o DM
• Exportaciones de seguidores y following
• Listas y comunidades

Para verificaciones recurrentes, programa flujos de Prefect o paginación REST en lugar de endpoints de monitor nativos. TwexAPI no expone trabajos de extracción CSV/JSON asíncronos como un producto separado.

Mantén las escrituras detrás de aprobación explícita. Guarda IDs confirmados antes de reintentar cualquier acción de escritura.

Fuente y próximos pasos

• Código fuente de prefect-x-api-scraper
• Horarios de Prefect
• Reintentos de tareas de Prefect
• Advanced Twitter Search
• Global Trending Tweets
• Agent MCP Handoff