Prefect

Prefect と TwexAPI 読み取り専用タスクで、スケジュールされた Twitter 検索、プロフィール、タイムライン、トレンドワークフローを Python で構築する。

Prefect は Python コード向けのオープンソースワークフローオーケストレーターです。prefect-x-api-scraper で繰り返し可能なツイート検索、プロフィール参照、タイムライン更新、トレンド確認を行えます。

コレクションは読み取り専用です。6つの非同期 Prefect タスクを提供し、各タスクは正規 TwexAPI JSON レスポンスを Python 辞書として返します。

リサーチ、エンリッチメント、ダッシュボード、アラート、インデックス作成にこのコレクションを使用してください。書き込み、フォロワーページネーション、エクスポートには直接 REST、Python SDK、または MCP を使用してください。

インストール

Python 3.10 以降を使用してください。

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

Prefect インストール後、認証情報ブロックを登録します。

prefect block register -m prefect_x_api_scraper

TwexAPI API キーを保存

TwexAPI ダッシュボード から API キーを作成し、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)

Prefect ブロックはフローとデプロイメント間で型付き設定を保存します。API キーをデプロイメント YAML、フローパラメータ、ログ、リポジトリに置かないでください。

適切な Prefect タスクを選択

Python で Twitter 自動化フローを構築

このフローは最近の投稿を検索し、ツイート行を正規化します。ID、著者、タイムスタンプ、メトリクス、URL、ページネーション状態を保持します。

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

asyncio で実行:

import asyncio

result = asyncio.run(social_signal_flow())

焦点を絞ったツイート検索クエリを記述

時系列モニタリングには sort_by="Latest" を使用します。エンゲージメント順発見には sort_by="Top" を使用します。ランキングは変わる可能性があるため、ツイート ID を永続化してください。

Twitter 検索パイプラインをスケジュール

.serve() で長時間実行ローカルプロセスを使用します。

from prefect.schedules import Cron


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

Docker、Kubernetes、サーバーレスワーカーにはワークプールデプロイメントを使用します。

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

ツイートとプロフィール結果をページネーション

カーソルページネーションはページ番号推測なしで大規模検索を継続します。元のリクエストは変更せず、返されたカーソルのみを渡します。

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

カーソルをデコードしないでください。不透明な文字列として扱います。各ページ永続化後にカーソルを保存してください。

スケジュール実行をべき等に

一時的障害のみリトライ

Prefect はリトライ遅延、ジッター、条件付きリトライをサポートします。無効な入力を変更なしでリトライしないでください。

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

ドキュメント化されたエラーをルーティング

完全な復旧ガイダンスは Error Handling と Rate Limits を参照してください。 | 5xx | 上限付きで一時的障害をリトライ。 |

MCP 計画代替

計画段階で Twexapi MCP を使用してエンドポイントを発見し、選択ルートを Prefect タスクまたは REST 呼び出しにコード化します。

推奨計画プロンプト:

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.

Prefect コレクションか直接 TwexAPI API か

6つのサポート読み取りには prefect-x-api-scraper を選択してください。ブロック、非同期呼び出し、検証、タスクメタデータを提供します。

以下には直接 REST、Python SDK、または MCP を使用してください。

• ツイート、返信、引用、いいね、フォロー、DM アクション
• フォロワーとフォローエクスポート
• リストとコミュニティ

定期チェックには Prefect フローまたは REST ページネーションをスケジュールしてください。TwexAPI はネイティブモニターエンドポイントを公開していません。TwexAPI は非同期 CSV/JSON 抽出ジョブを別製品として公開していません。

書き込みは明示的な承認の背後に置いてください。書き込みアクションをリトライする前に確認済み ID を保存してください。

ソースと次のステップ

• prefect-x-api-scraper source
• Prefect schedules
• Prefect task retries
• Advanced Twitter Search
• Global Trending Tweets
• Agent MCP Handoff