Prefect

Construisez des workflows planifiés de recherche Twitter, profil, timeline et tendances en Python avec Prefect et les tâches en lecture seule TwexAPI.

Prefect est un orchestrateur de workflow open source pour le code Python. Utilisez prefect-x-api-scraper pour des recherches de tweets reproductibles, lookups de profil, rafraîchissements de timeline et vérifications de tendances.

La collection est en lecture seule. Elle fournit six tâches Prefect asynchrones. Chaque tâche retourne la réponse JSON canonique TwexAPI en tant que dictionnaire Python.

Utilisez cette collection pour la recherche, l’enrichissement, les tableaux de bord, les alertes et l’indexation. Utilisez REST direct, le SDK Python ou MCP pour les écritures, la pagination d’abonnés et les exports.

Installation

Utilisez Python 3.10 ou plus récent.

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

Enregistrez le bloc d’identifiants après installation de Prefect :

prefect block register -m prefect_x_api_scraper

Stocker la clé API TwexAPI

Créez une clé API depuis le tableau de bord TwexAPI. Stockez-la dans un bloc 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)

Les blocs Prefect stockent la configuration typée entre flows et déploiements. Ne jamais placer des clés API dans le YAML de déploiement, les paramètres de flow, les logs ou les dépôts.

Choisir la bonne tâche Prefect

Construire un flow d’automatisation Twitter en Python

Ce flow recherche des posts récents et normalise les lignes de tweet. Il préserve les ID, auteurs, horodatages, métriques, URLs et l’état de pagination.

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"),
    }

Exécutez avec asyncio :

import asyncio

result = asyncio.run(social_signal_flow())

Écrire des requêtes de recherche de tweets ciblées

Utilisez sort_by="Latest" pour la surveillance chronologique. Utilisez sort_by="Top" pour la découverte classée par engagement. Persistez les ID de tweet car les classements peuvent changer.

Planifier le pipeline de recherche Twitter

Utilisez .serve() pour un processus local longue durée.

from prefect.schedules import Cron


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

Utilisez un déploiement work-pool pour les workers Docker, Kubernetes ou serverless.

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

Paginer les résultats de tweets et profils

La pagination par curseur continue les grandes recherches sans deviner les numéros de page. Gardez la requête originale inchangée. Passez uniquement le curseur retourné.

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

Ne décodez pas les curseurs. Traitez-les comme des chaînes opaques. Stockez chaque curseur après persistance de sa page.

Rendre les exécutions planifiées idempotentes

Réessayer uniquement les échecs transitoires

Prefect supporte les délais de retry, le jitter et les retries conditionnels. Ne pas réessayer des entrées invalides sans modification.

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],
)

Router les erreurs documentées

Consultez Gestion des erreurs et Limites de débit pour les conseils de récupération complets.

Alternative de planification MCP

Utilisez Twexapi MCP pendant la planification pour découvrir les endpoints, puis codifiez la route sélectionnée dans les tâches Prefect ou les appels REST.

Prompt de planification suggéré :

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.

Collection Prefect ou API TwexAPI directe

Choisissez prefect-x-api-scraper pour ses six lectures supportées. Il fournit des blocs, des appels async, la validation et les métadonnées de tâche.

Utilisez REST direct, le SDK Python ou MCP pour :

• Actions tweet, réponse, citation, like, follow ou DM
• Exports d’abonnés et following
• Listes et communautés

Pour les vérifications récurrentes, planifiez des flows Prefect ou la pagination REST plutôt que des endpoints de moniteur natifs. TwexAPI n’expose pas les jobs d’extraction CSV/JSON async comme produit séparé.

Gardez les écritures derrière une approbation explicite. Stockez les ID confirmés avant de réessayer une action d’écriture.

Source et prochaines étapes

• Source prefect-x-api-scraper
• Schedules Prefect
• Retries de tâches Prefect
• Advanced Twitter Search
• Global Trending Tweets
• Agent MCP Handoff