prefect-xquik 0.1.6

Prefect collection for scheduled X/Twitter data workflows with Xquik

Prefect tasks for Xquik X/Twitter data workflows

Prefect Guide · Releases · Xquik API Reference · Prefect Docs

Guide: https://docs.xquik.com/guides/prefect

Run scheduled X/Twitter data reads in Prefect 3 flows with a reusable XquikCredentials block, async tasks, retries, and normal Prefect deployment patterns.

Use prefect-xquik when a workflow needs public X/Twitter signals for research, monitoring, enrichment, dashboards, or alerts without maintaining scraper code in each flow.

What You Can Schedule

Workflow Need	Prefect Task	Xquik Endpoint
Search recent or top posts	search_tweets	GET /x/tweets/search
Look up a specific post	get_tweet	GET /x/tweets/{id}
Search public users	search_users	GET /x/users/search
Look up a user profile	get_user	GET /x/users/{id}
Fetch a user's timeline	get_user_tweets	GET /x/users/{id}/tweets
Read regional or global trends	get_trends	GET /x/trends

Install

pip install prefect-xquik

For local development:

uv sync
uv run ruff format --check .
uv run ruff check .
uv run pytest
uv build
uv run twine check dist/*

Register Blocks

prefect block register -m prefect_xquik

Create a block in the Prefect UI or with Python:

from prefect_xquik import XquikCredentials

credentials = XquikCredentials(api_key="<xquik-api-key>")
credentials.save("xquik", overwrite=True)

Store API keys in Prefect blocks, not in flow source files.

Example Flow

from prefect import flow
from prefect_xquik import XquikCredentials, get_trends, search_tweets


@flow
async def social_signal_flow() -> dict:
    credentials = XquikCredentials.load("xquik")

    tweets = await search_tweets(
        credentials,
        query='"prefect" OR "workflow orchestration"',
        query_type="Latest",
        limit=25,
    )
    trends = await get_trends(credentials, woeid=1, count=10)

    return {"tweets": tweets, "trends": trends}

Task Imports

from prefect_xquik import (
    get_trends,
    get_tweet,
    get_user,
    get_user_tweets,
    search_tweets,
    search_users,
)

Tasks return the raw Xquik JSON response as a Python dictionary. Configure Prefect runtime behavior with with_options:

from prefect_xquik import search_tweets

search_recent_tweets = search_tweets.with_options(
    name="Search Recent X Posts",
    retries=2,
    retry_delay_seconds=10,
)

API Contract

The credentials block sends x-api-key and the current xquik-api-contract header. The default contract is 2026-04-29, matching the public OpenAPI contract used to build this collection.

Documentation

• Xquik Prefect guide
• Xquik API reference
• Prefect integrations guide
• Prefect workflows and tasks

Development

uv sync
uv run ruff format --check .
uv run ruff check .
uv run pytest
uv build
uv run twine check dist/*

This repository follows Prefect's external collection layout so it can be moved under src/integrations/prefect-xquik later if Prefect maintainers request it.