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

<CardGroup cols={3}>
  <Card title="Tweet arama" icon="search">
    Anahtar kelime, hashtag, hesap, tarih ve X sorgu operatörleriyle arama yapın.
  </Card>

  <Card title="Tweet sorgulama" icon="message-square">
    Sayısal kimlikten tek bir herkese açık tweet getirin.
  </Card>

  <Card title="Profil arama" icon="users">
    Ad, kullanıcı adı veya konuya göre herkese açık X hesapları bulun.
  </Card>

  <Card title="Profil sorgulama" icon="user-round">
    Ekran adına göre tek bir herkese açık profil getirin.
  </Card>

  <Card title="Zaman çizelgelerini yenileme" icon="list">
    Kullanıcı zaman çizelgesi sayfasından son tweet'leri getirin.
  </Card>

  <Card title="Trend takibi" icon="trending-up">
    Ülke, konu ve içerik etiketine göre küresel trend tweet'lerini getirin.
  </Card>
</CardGroup>

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](/sdks/python) veya [MCP](/mcp/overview) kullanın.

## Kurulum

Python 3.10 veya daha yeni kullanın.

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

Prefect kurulumundan sonra kimlik bilgileri bloğunu kaydedin:

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

## TwexAPI API anahtarını saklayın

[TwexAPI dashboard](https://twexapi.io/dashboard) üzerinden bir API anahtarı oluşturun. `TwexApiCredentials` bloğu içinde saklayın.

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

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

<CardGroup cols={2}>
  <Card title="search_tweets" icon="search">
    `POST /twitter/advanced_search/page` çağırır. `query`, `cursor`, `sort_by` ve sayfalama alanlarını kabul eder.
  </Card>

  <Card title="get_tweet" icon="message-square">
    `POST /v2/tweet/detail` çağırır. Tek bir sayısal tweet kimliği geçirin.
  </Card>

  <Card title="search_users" icon="users">
    `GET /twitter/search-user/{keyword}/{target_count}` çağırır. Profil sorgusu kabul eder.
  </Card>

  <Card title="get_user" icon="user-round">
    `GET /twitter/{screen_name}/about` çağırır. `@` ile veya `@` olmadan ekran adlarını kabul eder.
  </Card>

  <Card title="get_user_tweets" icon="list">
    `POST /twitter/{screen_name}/timeline/page` çağırır. İmleç ve sayfa boyutunu destekler.
  </Card>

  <Card title="get_trends" icon="trending-up">
    `GET /twitter/global-trending/tweets` çağırır. `country`, `topic`, `content` ve `count` kabul eder.
  </Card>
</CardGroup>

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

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

asyncio ile çalıştırın:

```python
import asyncio

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

## Odaklı tweet arama sorguları yazın

<CardGroup cols={2}>
  <Card title="Tam ifade" icon="quote">
    Tam ifade için `"workflow orchestration"` kullanın.
  </Card>

  <Card title="Hesap filtresi" icon="at-sign">
    Tek hesaptan tweet'ler için `from:PrefectIO` kullanın.
  </Card>

  <Card title="Hashtag araması" icon="hash">
    Eşleşen hashtag'ler için `#prefect #python` kullanın.
  </Card>

  <Card title="Tarih penceresi" icon="calendar-range">
    Sorgu içinde `since:` ve `until:` tarihleri kullanın.
  </Card>

  <Card title="Yeniden gönderileri hariç tut" icon="repeat-2">
    Orijinal gönderiler önemliyse `-filter:retweets` kullanın.
  </Card>
</CardGroup>

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.

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

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

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

İ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

<CardGroup cols={2}>
  <Card title="Tweet kimliği" icon="message-square">
    Tweet satırlarını `tweet_id` ile upsert edin. Anahtar olarak metin kullanmayın.
  </Card>

  <Card title="Profil kimliği" icon="user-round">
    Profil satırlarını sayısal kullanıcı kimliğine göre upsert edin.
  </Card>

  <Card title="İmleç kontrol noktası" icon="list-tree">
    Her commit edilen sayfadan sonra `has_more` ve `next_cursor` saklayın.
  </Card>
</CardGroup>

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

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

## Belgelenmiş hataları yönlendirin

| Durum | Eylem |
| ------ | ------ |
| `400` | Eksik sorguları veya hatalı girdiyi düzeltin. Değiştirmeden yeniden denemeyin. |
| `401` | Kimlik bilgileri bloğundan geçerli bir API anahtarı yükleyin. |
| `403` | Sonraki çalıştırmadan önce hesap erişimini veya kredileri çözün. |
| `404` | Tweet kimliklerini, kullanıcı adlarını ve ekran adlarını kontrol edin. |
| `429` | Zamanlamayı yavaşlatın ve jitter'lı gecikmelerle yeniden deneyin. |
| `5xx` | Geçici hataları sınırlı geri çekilme ile yeniden deneyin. |

Tam kurtarma rehberi için [Error Handling](/guides/error-handling) ve [Rate Limits](/guides/rate-limits) bölümlerine bakın.

## MCP planlama alternatifi

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

Önerilen planlama prompt'u:

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

## 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](/sdks/python) 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](https://github.com/twexapi-dev/prefect-x-api-scraper)
* [Prefect schedules](https://docs.prefect.io/v3/concepts/schedules)
* [Prefect task retries](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)