Prefect

Prefect ve TwexAPI salt okunur görevleriyle Python'da zamanlanmış Twitter arama, profil, zaman çizelgesi ve trend iş akışları oluşturun.

Prefect, Python kodu için açık kaynaklı bir iş akışı orkestratörüdür. Tekrarlanabilir tweet aramaları, profil sorgulamaları, zaman çizelgesi yenilemeleri ve trend kontrolleri için prefect-x-api-scraper kullanın.

Koleksiyon salt okunurdur. Altı asenkron Prefect görevi sağlar. Her görev kanonik TwexAPI JSON yanıtını Python sözlüğü olarak döndürür.

Bu koleksiyonu araştırma, zenginleştirme, panolar, uyarılar ve indeksleme için kullanın. Yazma, takipçi sayfalama ve dışa aktarımlar için doğrudan REST, Python SDK veya MCP kullanın.

Kurulum

Python 3.10 veya daha yeni kullanın.

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

Prefect kurulumundan sonra kimlik bilgileri bloğunu kaydedin:

prefect block register -m prefect_x_api_scraper

TwexAPI API anahtarını saklayın

TwexAPI dashboard üzerinden bir API anahtarı oluşturun. TwexApiCredentials bloğu içinde saklayın.

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 blokları, akışlar ve dağıtımlar arasında tipli yapılandırma saklar. API anahtarlarını dağıtım YAML’ına, akış parametrelerine, günlüklere veya depolara asla koymayın.

Doğru Prefect görevini seçin

Python’da Twitter otomasyon akışı oluşturun

Bu akış son gönderileri arar ve tweet satırlarını normalleştirir. Kimlikleri, yazarları, zaman damgalarını, metrikleri, URL’leri ve sayfalama durumunu korur.

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 ile çalıştırın:

import asyncio

result = asyncio.run(social_signal_flow())

Odaklı tweet arama sorguları yazın

Kronolojik izleme için sort_by="Latest" kullanın. Etkileşim sıralı keşif için sort_by="Top" kullanın. Sıralamalar değişebileceği için tweet kimliklerini kalıcı hale getirin.

Twitter arama hattını zamanlayın

Uzun süreli yerel süreç için .serve() kullanın.

from prefect.schedules import Cron


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

Docker, Kubernetes veya sunucusuz worker’lar için work-pool dağıtımı kullanın.

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

Tweet ve profil sonuçlarını sayfalayın

İmleç sayfalama, sayfa numaralarını tahmin etmeden büyük aramaları sürdürür. Orijinal isteği değiştirmeden tutun. Yalnızca dönen imleci geçirin.

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

İmleçleri çözmeyin. Opak dizgiler olarak ele alın. Her imleci sayfasını kalıcı hale getirdikten sonra saklayın.

Zamanlanmış çalıştırmaları idempotent yapın

Yalnızca geçici hataları yeniden deneyin

Prefect yeniden deneme gecikmeleri, jitter ve koşullu yeniden denemeleri destekler. Geçersiz girdileri değiştirmeden yeniden denemeyin.

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

Belgelenmiş hataları yönlendirin

Tam kurtarma rehberi için Error Handling ve Rate Limits bölümlerine bakın.

MCP planlama alternatifi

Uç noktaları keşfetmek için planlama sırasında Twexapi MCP kullanın, ardından seçilen rotayı Prefect görevlerinde veya REST çağrılarında kodlayın.

Önerilen planlama prompt’u:

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 koleksiyonu mu doğrudan TwexAPI API mi?

Altı desteklenen okuma için prefect-x-api-scraper seçin. Bloklar, asenkron çağrılar, doğrulama ve görev meta verileri sağlar.

Şunlar için doğrudan REST, Python SDK veya MCP kullanın:

• Tweet, yanıt, alıntı, beğeni, takip veya DM eylemleri
• Takipçi ve takip dışa aktarımları
• Listeler ve topluluklar

Tekrarlayan kontroller için yerel monitör uç noktaları yerine Prefect akışlarını veya REST sayfalamasını zamanlayın. TwexAPI asenkron CSV/JSON çıkarma işlerini ayrı bir ürün olarak sunmaz.

Yazma işlemlerini açık onayın arkasında tutun. Herhangi bir yazma işlemini yeniden denemeden önce onaylanmış kimlikleri saklayın.

Kaynak ve sonraki adımlar

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