Prefect

pip install \
  https://github.com/Xquik-dev/prefect-xquik/releases/download/v0.1.5/prefect_xquik-0.1.5-py3-none-any.whl

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",
)
credentials.save("xquik", overwrite=True)

from __future__ import annotations

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


@flow
async def social_signal_flow() -> dict[str, object]:
    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}


if __name__ == "__main__":
    import asyncio

    asyncio.run(social_signal_flow())

python social_signal_flow.py

prefect deploy social_signal_flow.py:social_signal_flow --name xquik-social-signals

from prefect_xquik import search_tweets

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

page = await search_recent_tweets(
    credentials,
    query="workflow orchestration lang:en",
    query_type="Latest",
    limit=100,
)

first_page = await search_tweets(
    credentials,
    query="prefect lang:en",
    query_type="Latest",
    limit=100,
)

next_cursor = first_page.get("next_cursor") or first_page.get("nextCursor")

if next_cursor:
    second_page = await search_tweets(
        credentials,
        query="prefect lang:en",
        query_type="Latest",
        cursor=str(next_cursor),
        limit=100,
    )