> ## Documentation Index
> Fetch the complete documentation index at: https://docs.xquik.com/llms.txt
> Use this file to discover all available pages before exploring further.
# Twitter search pipeline with Python & Prefect
> Build scheduled Twitter search, profile, timeline, and trend workflows in Python with Prefect. Paginate tweet and user pages, then retry failures. See examples.
For the complete documentation index, see llms.txt.
Prefect is an open-source workflow orchestrator for Python code. This Prefect Python tutorial schedules Twitter API Python reads through Xquik. It builds a Prefect Python workflow orchestration pipeline for tweets, profiles, timelines, and trends.
Use [`prefect-xquik`](https://github.com/Xquik-dev/prefect-xquik) for repeatable tweet searches, profile lookups, timeline refreshes, and trend checks. Use it for scheduled Twitter API reads in Python. This Prefect Python library gives you typed credentials and 6 read-only tasks. This collection is not a general Prefect Python SDK.
The collection is read-only. It provides 6 asynchronous Prefect tasks. Each task returns the canonical Xquik JSON response as a Python dictionary.
You can track Python functions as data pipelines with Prefect. Prefect flows wrap normal Python code with retries, schedules, and state tracking. Prefect offers run history, deployment controls, and execution environments for each data workflow. The Prefect UI shows every created task and flow state in real time.
Search keywords, hashtags, accounts, dates, and X query operators.
Fetch one public tweet from its numeric ID.
Find public X accounts by name, username, or topic.
Retrieve one public profile by username or user ID.
Fetch recent tweets with optional replies and parent context.
Retrieve worldwide or regional trending topics by WOEID.
Use this collection for research, enrichment, dashboards, alerts, and indexing. Use direct REST, SDK, or MCP calls for writes, monitors, webhooks, and exports.
## Install a verified Prefect release
Use Python 3.10 or newer. Pin the verified source tag and Prefect release.
```bash theme={null}
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"
```
You can install `prefect-xquik` 0.1.8 or earlier releases from PyPI. Use the tested `v0.1.7` source pin with Prefect 3. It also uses the canonical Xquik API base URL. Tag `v0.1.8` only changes package maintenance. It also requires newer Prefect releases.
The explicit `importlib-metadata` pin fixes Prefect's clean Python 3.12 CLI import. Prefect documents the upstream issue in [#22011](https://github.com/PrefectHQ/prefect/issues/22011) and [#22137](https://github.com/PrefectHQ/prefect/issues/22137).
Use the tested pins above for production. Use a separate sandbox for compatibility checks.
```bash theme={null}
python -m venv .venv-prefect-compat
source .venv-prefect-compat/bin/activate
python -m pip install -U prefect prefect-xquik
prefect server start
```
Keep the server running. Open another shell and activate the same environment.
```bash theme={null}
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
```
PowerShell users should activate `.venv-prefect-compat\Scripts\Activate.ps1` instead. Run `pip install -U prefect` only inside this sandbox. Run `prefect server start` before registering the block.
Keep the tested pins above for this integration. Treat Prefect and the collection as separate Python packages.
## Store the Xquik API key
Create an [Xquik API key](/x-api-quickstart). Store it inside an `XquikCredentials` block.
```python theme={null}
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)
```
Prefect blocks store typed configuration across flows and deployments. The API key uses Pydantic `SecretStr`. Prefect hides its value in normal block rendering.
Never place API keys in deployment YAML, flow parameters, logs, or repositories. Limit block access to deployments that require Xquik reads.
## Understand the Prefect Python runtime
Prefect runs ordinary Python functions. The `@flow` decorator creates a tracked flow run. The `@task` decorator creates task state with optional retries. Configure retries per task or through a nonzero global default.
The `from prefect import flow, task` statement imports both decorators.
A dynamic workflow can branch from returned tweets, profiles, or trends. Prefect does not require YAML files for Python flows. Use Python code for branches, loops, and task dependencies.
Store the local profile name or server URL in environment variables. Store Xquik keys in `XquikCredentials` for deployed flows.
## Choose the right Prefect task
Calls `GET /x/tweets/search`. Accepts `query`, `cursor`, `limit`, `query_type`, `since_time`, and `until_time`.
Calls `GET /x/tweets/{id}`. Pass one numeric tweet ID.
Calls `GET /x/users/search`. Accepts a profile query and optional cursor.
Calls `GET /x/users/{id}`. Accepts usernames with or without `@`.
Calls `GET /x/users/{id}/tweets`. Supports replies, parent tweets, and cursors.
Calls `GET /x/trends`. Accepts `woeid` and a count from 1 through 50.
`search_tweets` limits each request to 200 tweets. `get_trends` limits each request to 50 topics. The collection validates these ranges before sending a request.
## Build a Twitter automation flow in Python
This flow searches recent Prefect posts and normalizes tweet rows. It keeps IDs, authors, timestamps, metrics, URLs, and pagination state.
```python theme={null}
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"),
}
```
Run it with explicit ISO 8601 boundaries.
```python theme={null}
import asyncio
result = asyncio.run(
social_signal_flow(
since_time="2026-08-01T00:00:00Z",
until_time="2026-08-02T00:00:00Z",
)
)
```
The timestamps define a reproducible search window. Store them beside every run checkpoint.
## Write focused tweet search queries
Narrow each search to avoid irrelevant rows.
Use `"workflow orchestration"` for an exact phrase.
Use `from:PrefectIO` for tweets from one account.
Use `#prefect #python` for matching hashtags.
Use `prefect min_faves:10` for a like threshold.
Use `since:` and `until:` dates inside a query.
Use `-filter:retweets` when original posts matter.
Use `query_type="Latest"` for chronological monitoring. Use `query_type="Top"` for engagement-ranked discovery. Rankings can change. Store tweet IDs.
## Schedule the Twitter search pipeline
Use `.serve()` for a long-running local process. Add a cron schedule and timezone.
```python theme={null}
from prefect.schedules import Cron
if __name__ == "__main__":
social_signal_flow.serve(
name="xquik-social-signals",
schedule=Cron("0 * * * *", timezone="UTC"),
)
```
The scheduled flow derives the previous hour on every run. Pass explicit boundaries for backfills.
Use a work-pool deployment for Docker, Kubernetes, or serverless workers.
```bash theme={null}
prefect deploy social_signal_flow.py:social_signal_flow \
--name xquik-social-signals \
--pool production
```
Prefect schedules create flow runs. Workers execute those runs. Confirm that the selected work pool has an active worker.
## Paginate tweet and profile results
Cursor pagination continues large searches without guessing page numbers. Keep the original request unchanged. Pass only the returned cursor.
```python theme={null}
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())
```
Do not decode cursors. Treat them as opaque strings. Store each cursor after storing its page. Resume from that checkpoint after a worker restart.
Use the same pattern with `search_users` and `get_user_tweets`. Keep `query`, `query_type`, `limit`, replies, and timestamps unchanged.
## Make scheduled runs idempotent
Scheduled windows can overlap. Workers can also retry completed requests. Prevent duplicate downstream rows with stable identifiers.
Upsert tweet rows by `id`. Do not use text as a key.
Upsert profile rows by numeric user `id`.
Store `since_time`, `until_time`, query, and ordering.
Store `has_next_page` and `next_cursor` after each committed page.
Keep tweet rows, profile rows, trend rows, and checkpoints separate. This simplifies retries and schema changes.
## Retry only transient failures
Prefect supports retry delays, jitter, and conditional retries. Do not retry invalid inputs or billing failures unchanged.
```python theme={null}
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,
)
```
Add jitter so scheduled tasks avoid synchronized retries. The collection does not expose response headers on `XquikError`. Use conservative delays for `429`, or call REST directly when header-aware handling matters.
## Route every documented error
`XquikError` exposes a sanitized message, optional `status_code`, and raw `response_text`. Do not store the raw response. Store the status, task run ID, endpoint, and safe error category.
Fix missing queries, invalid limits, or malformed input. Do not retry unchanged.
Load a valid Xquik API key from the credentials block.
Resolve subscription or credits before the next scheduled run.
Check tweet IDs, usernames, and user IDs. Search routes omit this status.
Retry with bounded backoff. Stop after the configured attempt cap.
Slow the schedule and retry with jittered delays.
Retry transient X retrieval failures with a cap.
Retry bounded connection failures when `status_code` is absent.
These statuses cover all 6 collection endpoints. Do not add undocumented statuses to route-specific retry rules.
## Control concurrency and rate limits
Multiple schedules can overlap and share one API key. Create one global rate limit with slot decay.
```bash theme={null}
prefect gcl create xquik-read-rate --limit 500 --slot-decay-per-second 500
```
Acquire one shared slot before every Xquik read.
```python theme={null}
from prefect.concurrency.asyncio import rate_limit
async def acquire_xquik_read_slot() -> None:
await rate_limit("xquik-read-rate", occupy=1, strict=True)
```
Await this helper before `search_tweets`, `get_tweet`, and `search_users`. Also await it before `get_user`, `get_user_tweets`, and `get_trends`.
Use a separate Prefect concurrency limit for in-flight task runs. Concurrency limits only cap active work. They do not set request frequency. Keep retry policies aligned with [Xquik rate-limit guidance](/guides/rate-limits).
Read endpoints share a 500 per 1s user bucket. Treat this as one shared capacity limit.
Use one limit across tweet search, profile lookup, timelines, and trends. Make every task use that same account limit. Slow low-priority refreshes before delaying interactive reads.
## Build profile and timeline workflows
Use `search_users` when a name or topic can match multiple accounts. Use `get_user` when you already know the username or user ID.
`get_user_tweets` retrieves one account's recent timeline. Set `include_replies=True` for conversations. Set `include_parent_tweet=True` when reply context matters.
Normalize profiles into stable fields:
* `id`
* `username`
* `name`
* `followers`
* `following`
* `verified`
* `profilePicture`
Normalize tweets separately. Keep `id`, `text`, `author`, `createdAt`, metrics, and `url`.
## Build regional trend alerts
`get_trends(credentials, woeid=1, count=30)` returns worldwide trends. Pass another valid WOEID for a region.
Store each trend's `name`, `rank`, `query`, and `description` when present. Store response `woeid` and `count` with the alert batch.
Trends are discovery signals, not verified facts. Validate important topics through tweet search before alerting customers.
## Result handoff
Store `tweets`, `has_next_page`, and `next_cursor`. Normalize tweets before loading a warehouse or dashboard.
Store `users`, `has_next_page`, and `next_cursor`. Keep numeric user IDs as primary keys.
Store `trends`, `count`, and `woeid`. Keep each trend's rank.
Store status, endpoint, task run ID, attempt count, and safe error category.
Do not pass raw dictionaries directly into alerts or generators. Define compact row contracts first. Keep row fields stable when API responses add fields.
## Prefect collection or direct Xquik API
Choose `prefect-xquik` for its 6 supported reads. It provides blocks, async calls, validation, and task metadata.
Use direct REST, a generated SDK, or MCP for:
* Tweet, reply, quote, repost, like, follow, or DM actions
* Follower and following exports
* Tweet replies, quotes, reposts, lists, and communities
* Monitors, signed webhooks, and stored event replay
* CSV, JSON, XLSX, Markdown, or PDF extraction jobs
* Request filters absent from the 6 Prefect tasks
Keep writes behind explicit approval. Store confirmed IDs before retrying any write action.
## Common Prefect Twitter API questions
### What is Prefect Python?
Prefect orchestrates Python functions as tracked flows and tasks. It adds schedules, retries, state, logs, and deployment controls without replacing Python syntax.
### What does this Python Prefect tutorial cover?
It builds a Prefect Python pipeline for scheduled Twitter searches and timeline reads. The flow keeps tweets, cursors, time windows, and documented errors.
### How does a Twitter search API Python flow work?
The flow calls `search_tweets` with a query and time window. It normalizes tweet IDs, text, authors, metrics, URLs, and pagination fields.
### Is this Twitter automation Python workflow read-only?
Yes. The Prefect Python API collection reads tweets, profiles, timelines, and trends. Send approved writes through separate REST, SDK, or MCP routes.
### Is `prefect-xquik` a Prefect Python SDK?
No. You get 6 asynchronous read tasks. Use Xquik SDKs when a workflow needs broader REST coverage.
### Which Prefect Python version should I install?
Use Python 3.10 or newer with the tested Prefect 3.4.25 pin. Validate newer Prefect releases before changing production dependencies.
### Where is the Prefect Python GitHub collection?
The [prefect-xquik repository](https://github.com/Xquik-dev/prefect-xquik) contains the package, examples, tests, and release tags. Check source changes before upgrading your pinned deployment.
### Python Prefect vs airflow: which fits Twitter search?
Pick Prefect when Python control flow and task retries matter. Choose Airflow when your team already runs DAG-based schedules. Either system still needs cursor, idempotency, and rate-limit controls.
### How do I automate Twitter search with Python?
Install `prefect-xquik`, register `XquikCredentials`, then schedule `search_tweets`. Use explicit time windows and cursor checkpoints.
### Can Prefect schedule Twitter profile and timeline reads?
Yes. Use `get_user` for profiles and `get_user_tweets` for recent timelines. Enable replies only when needed.
### Can the Prefect collection post tweets?
No. The 6 tasks are read-only. Use Xquik REST, SDK, or MCP routes for approved writes.
### How do I handle Twitter API rate limits in Prefect?
Await the shared rate limiter before all 6 tasks. Retry `429` with jittered delays and a hard attempt cap. Keep concurrency limits separate.
### How do I prevent duplicate tweets across scheduled runs?
Enforce a unique tweet ID constraint. Upsert rows or ignore conflicts during overlapping runs. Save time windows and cursors after each committed page.
### Should I use Prefect or cron for Twitter automation?
Use cron for one simple local script. Use Prefect for retries, blocks, schedules, workers, run history, and deployment controls.
### Where should I store the Xquik API key?
Store it in an `XquikCredentials` block. Never pass it as a flow parameter.
## Source and next steps
* [prefect-xquik source](https://github.com/Xquik-dev/prefect-xquik)
* [prefect-xquik v0.1.7](https://github.com/Xquik-dev/prefect-xquik/releases/tag/v0.1.7)
* [prefect-xquik on PyPI](https://pypi.org/project/prefect-xquik/)
* [Prefect schedules](https://docs.prefect.io/v3/concepts/schedules)
* [Prefect task retries](https://docs.prefect.io/v3/how-to-guides/workflows/retries)
* [Tweet Search API](/api-reference/x/search-tweets)
* [User Timeline API](/api-reference/x/user-tweets)
* [Rate Limits](/guides/rate-limits)
* [Error Handling](/guides/error-handling)
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.