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:
- Getting Started - Installation and basic usage
- Query Options - All configuration options explained
- Mutations - Handling POST/PUT/DELETE operations
- Cache Management - Invalidation, prefetching, manual updates
- Observers - Reactive state updates
- Advanced Patterns - Dependent queries, parallel fetching
- Framework Integrations - FastAPI, Tkinter, Textual, CLI tools, Jupyter
- API Reference - Complete API 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)
| File | Size | Uploaded | |
|---|---|---|---|
| pystackquery-1.0.2.tar.gz | 95.0 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| 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
|