Twitter Search Pipeline with Python & Prefect

python -m pip install \
  "prefect==3.4.25" \
  "importlib-metadata==8.7.0" \
  "prefect-xquik @ https://github.com/Xquik-dev/prefect-xquik/archive/refs/tags/v0.1.7.tar.gz"

python -m venv .venv-prefect-compat
source .venv-prefect-compat/bin/activate
python -m pip install -U prefect prefect-xquik
prefect server start

source .venv-prefect-compat/bin/activate
prefect config set PREFECT_API_URL="http://127.0.0.1:4200/api"
prefect block register -m prefect_xquik

from prefect_xquik import XquikCredentials

credentials = XquikCredentials(
    api_key="xq_YOUR_KEY_HERE",
    base_url="https://xquik.com/api/v1",
    api_contract="2026-04-29",
    timeout_seconds=30,
)
credentials.save("xquik", overwrite=True)

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from typing import Any, Optional

from prefect import flow, task
from prefect_xquik import XquikCredentials, 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("tweets", []):
        if not isinstance(tweet, dict):
            continue
        author = tweet.get("author")
        rows.append(
            {
                "tweet_id": tweet.get("id"),
                "text": tweet.get("text"),
                "created_at": tweet.get("createdAt"),
                "url": tweet.get("url"),
                "author": author if isinstance(author, dict) else {},
                "like_count": tweet.get("likeCount"),
                "repost_count": tweet.get("retweetCount"),
                "reply_count": tweet.get("replyCount"),
            }
        )
    return rows


@flow(name="Xquik Twitter Search")
async def social_signal_flow(
    since_time: Optional[str] = None,
    until_time: Optional[str] = None,
) -> dict[str, Any]:
    if (since_time is None) != (until_time is None):
        raise ValueError("Pass both time boundaries or neither boundary")
    if since_time is None and until_time is None:
        window_end = datetime.now(timezone.utc)
        window_start = window_end - timedelta(hours=1)
        since_time = window_start.isoformat().replace("+00:00", "Z")
        until_time = window_end.isoformat().replace("+00:00", "Z")

    credentials = XquikCredentials.load("xquik")
    page = await search_tweets(
        credentials,
        query='"workflow orchestration" lang:en -filter:retweets',
        query_type="Latest",
        since_time=since_time,
        until_time=until_time,
        limit=100,
    )
    rows = await normalize_tweet_page(page)
    return {
        "tweet_rows": rows,
        "has_more": page.get("has_next_page", False),
        "next_cursor": page.get("next_cursor"),
    }

import asyncio

result = asyncio.run(
    social_signal_flow(
        since_time="2026-08-01T00:00:00Z",
        until_time="2026-08-02T00:00:00Z",
    )
)

from prefect.schedules import Cron


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

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

from typing import Any, Optional

from prefect import flow
from prefect_xquik import XquikCredentials, search_tweets


@flow
async def collect_tweet_pages(query: str) -> list[dict[str, Any]]:
    credentials = XquikCredentials.load("xquik")
    rows_by_id: dict[str, dict[str, Any]] = {}
    cursor: Optional[str] = None

    while True:
        page = await search_tweets(
            credentials,
            query=query,
            query_type="Latest",
            limit=100,
            cursor=cursor,
        )

        for tweet in page.get("tweets", []):
            if isinstance(tweet, dict) and tweet.get("id"):
                rows_by_id[str(tweet["id"])] = tweet

        cursor_value = page.get("next_cursor")
        has_more = bool(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())

from prefect_xquik import XquikError, search_tweets


def retry_transient_xquik_error(_task, _task_run, state) -> bool:
    try:
        state.result()
    except XquikError as error:
        return error.status_code is None or error.status_code in {424, 429, 502}
    return False


search_recent_tweets = search_tweets.with_options(
    name="Search Recent Tweets",
    retries=4,
    retry_delay_seconds=[5, 15, 45, 120],
    retry_jitter_factor=0.5,
    retry_condition_fn=retry_transient_xquik_error,
)

prefect gcl create xquik-read-rate --limit 300 --slot-decay-per-second 300

from prefect.concurrency.asyncio import rate_limit


async def acquire_xquik_read_slot() -> None:
    await rate_limit("xquik-read-rate", occupy=1, strict=True)