Skip to main content

PyStackQuery

Async data fetching and caching library for Python.

PyStackQuery handles the hard parts of working with async data: caching, deduplication, retries, and reactive state updates. You focus on what to fetch, the library handles how.

Installation

pip install pystackquery

Requires Python 3.11+

Quick Start

import asyncio
from pystackquery import QueryClient, QueryOptions

client = QueryClient()

async def fetch_user(user_id: int) -> dict:
    # Your async fetch logic here
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/users/{user_id}") as resp:
            return await resp.json()

async def main():
    # Fetch with automatic caching
    user = await client.fetch_query(
        QueryOptions(
            query_key=("user", "123"),
            query_fn=lambda: fetch_user(123)
        )
    )

    # Second call returns cached data instantly
    user_again = await client.fetch_query(
        QueryOptions(
            query_key=("user", "123"),
            query_fn=lambda: fetch_user(123)
        )
    )

asyncio.run(main())

That's it. The first call fetches from the API. The second call returns instantly from cache.

What Problems Does This Solve?

Without PyStackQuery, you write code like this over and over:

cache = {}
pending = {}
lock = asyncio.Lock()

async def get_user(user_id):
    key = f"user_{user_id}"

    async with lock:
        if key in cache:
            return cache[key]
        if key in pending:
            return await pending[key]

        task = asyncio.create_task(fetch_user(user_id))
        pending[key] = task

    try:
        result = await task
        cache[key] = result
        return result
    finally:
        del pending[key]

With PyStackQuery, you write:

user = await client.fetch_query(
    QueryOptions(("user", user_id), lambda: fetch_user(user_id))
)

The library handles:

  • Caching
  • Request deduplication (concurrent calls share one request)
  • Automatic retries with backoff
  • Stale-while-revalidate
  • Cache invalidation
  • Reactive updates

Core Concepts

Query Keys

Every query needs a unique key. Keys are tuples of strings:

("users",)                    # All users
("user", "123")               # Specific user
("posts", "user", "123")      # Posts by user 123

Keys enable:

  • Cache lookups
  • Partial invalidation (invalidate ("users",) clears all user queries)

Query Options

Configure how a query behaves:

QueryOptions(
    query_key=("user", "123"),
    query_fn=lambda: fetch_user(123),
    stale_time=60.0,    # Data fresh for 60 seconds
    retry=3,            # Retry 3 times on failure
)

Stale-While-Revalidate

When data becomes stale, you get the cached data immediately while a background refresh happens:

# First call: fetches from API
data = await client.fetch_query(opts)

# Wait for stale_time to pass...

# Second call: returns stale data instantly, refreshes in background
data = await client.fetch_query(opts)

Your users see data immediately. Fresh data loads in the background.

Documentation

See the docs/ folder for comprehensive documentation:

License

MIT

Release files for pystackquery 1.0.2

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for pystackquery 1.0.2
File Size Uploaded
pystackquery-1.0.2.tar.gz 95.0 kB Details

Built distribution (wheel)

Table of built distributions (wheels) for pystackquery 1.0.2
File Interpreter ABI Platform
pystackquery-1.0.2-py3-none-any.whl Python 3 none any Details

Total release size: 114.8 kB

Release files / pystackquery-1.0.2.tar.gz

Download URL pystackquery-1.0.2.tar.gz
Size 95.0 kB
Tags Source
SHA-256 checksum
How to use checksums
43a49c0baa9887e230eb7aec1c448dbc5deb22545900ad618aead841f7d08ff0
BLAKE2b-256 checksum
How to use checksums
e0b0d7a2162020f3b9b0c929610b49e23c955713fd1ab3002b01163c319dcd89
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/6.1.0 CPython/3.13.7

Release files / pystackquery-1.0.2-py3-none-any.whl

Download URL pystackquery-1.0.2-py3-none-any.whl
Size 19.8 kB
Tags Python 3
SHA-256 checksum
How to use checksums
26708acac47675747546b104ff0574001aaed541442f6bae67015b7df5ae2b81
BLAKE2b-256 checksum
How to use checksums
92d014435c64c654cbca80526eae84cd48f8183411808cf9b3f93f7218ab1664
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/6.1.0 CPython/3.13.7

Release history Release notifications | RSS feed

This release

1.0.2 This release

2 release files

1.0.1

2 release files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page