---
title: "Prefect"
description: "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`](https://github.com/twexapi-dev/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.

<CardGroup cols={3}>
  <Card title="Rechercher des tweets" icon="search">
    Recherchez des mots-clés, hashtags, comptes, dates et opérateurs de requête X.
  </Card>

  <Card title="Lookup de tweets" icon="message-square">
    Récupérez un tweet public depuis son ID numérique.
  </Card>

  <Card title="Rechercher des profils" icon="users">
    Trouvez des comptes X publics par nom, username ou sujet.
  </Card>

  <Card title="Lookup de profils" icon="user-round">
    Récupérez un profil public par nom d'écran.
  </Card>

  <Card title="Rafraîchir les timelines" icon="list">
    Récupérez les tweets récents d'une page de timeline utilisateur.
  </Card>

  <Card title="Suivre les tendances" icon="trending-up">
    Récupérez les tweets tendance globaux par pays, sujet et tag de contenu.
  </Card>
</CardGroup>

Utilisez cette collection pour la recherche, l'enrichissement, les tableaux de bord, les alertes et l'indexation. Utilisez REST direct, le [SDK Python](/sdks/python) ou [MCP](/mcp/overview) pour les écritures, la pagination d'abonnés et les exports.

## Installation

Utilisez Python 3.10 ou plus récent.

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

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

```bash
prefect block register -m prefect_x_api_scraper
```

## Stocker la clé API TwexAPI

Créez une clé API depuis le [tableau de bord TwexAPI](https://twexapi.io/dashboard). Stockez-la dans un bloc `TwexApiCredentials`.

```python
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

<CardGroup cols={2}>
  <Card title="search_tweets" icon="search">
    Appelle `POST /twitter/advanced_search/page`. Accepte `query`, `cursor`, `sort_by` et les champs de pagination.
  </Card>

  <Card title="get_tweet" icon="message-square">
    Appelle `POST /v2/tweet/detail`. Passez un ID de tweet numérique.
  </Card>

  <Card title="search_users" icon="users">
    Appelle `GET /twitter/search-user/{keyword}/{target_count}`. Accepte une requête de profil.
  </Card>

  <Card title="get_user" icon="user-round">
    Appelle `GET /twitter/{screen_name}/about`. Accepte des noms d'écran avec ou sans `@`.
  </Card>

  <Card title="get_user_tweets" icon="list">
    Appelle `POST /twitter/{screen_name}/timeline/page`. Supporte les curseurs et la taille de page.
  </Card>

  <Card title="get_trends" icon="trending-up">
    Appelle `GET /twitter/global-trending/tweets`. Accepte `country`, `topic`, `content` et `count`.
  </Card>
</CardGroup>

## 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.

```python
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 :

```python
import asyncio

result = asyncio.run(social_signal_flow())
```

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

<CardGroup cols={2}>
  <Card title="Phrase exacte" icon="quote">
    Utilisez `"workflow orchestration"` pour une phrase exacte.
  </Card>

  <Card title="Filtre de compte" icon="at-sign">
    Utilisez `from:PrefectIO` pour les tweets d'un compte.
  </Card>

  <Card title="Recherche par hashtag" icon="hash">
    Utilisez `#prefect #python` pour les hashtags correspondants.
  </Card>

  <Card title="Plage de dates" icon="calendar-range">
    Utilisez les dates `since:` et `until:` dans une requête.
  </Card>

  <Card title="Exclure les republications" icon="repeat-2">
    Utilisez `-filter:retweets` lorsque les posts originaux comptent.
  </Card>
</CardGroup>

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.

```python
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.

```bash
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é.

```python
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

<CardGroup cols={2}>
  <Card title="Identité tweet" icon="message-square">
    Upsert les lignes de tweet par `tweet_id`. Ne pas utiliser le texte comme clé.
  </Card>

  <Card title="Identité profil" icon="user-round">
    Upsert les lignes de profil par ID utilisateur numérique.
  </Card>

  <Card title="Checkpoint curseur" icon="list-tree">
    Stockez `has_more` et `next_cursor` après chaque page commitée.
  </Card>
</CardGroup>

## 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.

```python
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

| Statut | Action |
| ------ | ------ |
| `400` | Corrigez les requêtes manquantes ou l'entrée mal formée. Ne pas réessayer sans modification. |
| `401` | Chargez une clé API valide depuis le bloc d'identifiants. |
| `403` | Résolvez l'accès au compte ou les crédits avant la prochaine exécution. |
| `404` | Vérifiez les ID de tweet, usernames et noms d'écran. |
| `429` | Ralentissez le planning et réessayez avec des délais jitterés. |
| `5xx` | Réessayez les échecs transitoires avec un plafond. |

Consultez [Gestion des erreurs](/guides/error-handling) et [Limites de débit](/guides/rate-limits) pour les conseils de récupération complets.

## Alternative de planification MCP

Utilisez [Twexapi MCP](/mcp/overview) 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é :

```txt
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](/sdks/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](https://github.com/twexapi-dev/prefect-x-api-scraper)
* [Schedules Prefect](https://docs.prefect.io/v3/concepts/schedules)
* [Retries de tâches Prefect](https://docs.prefect.io/v3/how-to-guides/workflows/retries)
* [Advanced Twitter Search](/api-reference/search-endpoints/get-data-page-twitter-advanced-search-page-post)
* [Global Trending Tweets](/api-reference/trending-endpoints/get-global-trending-tweets-api-twitter-global-trending-tweets-get)
* [Agent MCP Handoff](/mcp/agent-handoff)