Skip to content

Repository files navigation

CacheFlight

In-process async request deduplication for Python.

Prevent duplicate expensive async work under concurrent load. When many callers request the same key simultaneously, CacheFlight ensures the task executes once and shares the result with all waiters.

v1 scope: single-process deduplication only. This is not a distributed lock or a persistent cache.

How It Works

Without CacheFlight, N concurrent callers for the same resource each independently start their own task:

Caller A ──► task() ──► result
Caller B ──► task() ──► result (duplicate work!)
Caller C ──► task() ──► result (duplicate work!)

With CacheFlight, the first caller starts the task; all others join as waiters. All receive the same result when it resolves:

Caller A ──► task() ─────────────► result
Caller B ──► (waits) ──────────► same result (no duplicate work)
Caller C ──► (waits) ──────────► same result (no duplicate work)

The in-flight entry is cleaned up immediately after the task resolves or rejects — CacheFlight stores nothing at rest. It is purely a concurrency coalescer, not a cache.

Installation

pip install cacheflight

Setup & Try the Examples

1. Clone and install

git clone https://github.com/cacheflight/cacheflight-python.git
cd cacheflight-python
python -m venv .venv
# Windows
.venv\Scripts\activate
# macOS / Linuxsource .venv/bin/activate
pip install -e ".[dev]"

2. Run the tests

pytest -v

3. Try the FastAPI example

pip install fastapi uvicorn
uvicorn examples.fastapi_example:app --reload

Then fire concurrent requests to see deduplication in action:

# Linux / macOSforiin$(seq 1 50);do curl -s http://localhost:8000/user/42 &done;wait# PowerShell
1..50 | ForEach-Object -Parallel { Invoke-RestMethod http://localhost:8000/user/42 }

Check /statstotal_db_calls will be 1 regardless of how many requests were fired.

4. Try the aiohttp example

pip install aiohttp
python examples/aiohttp_example.py
# Server runs on http://localhost:8080

5. Run the benchmarks

python -m benchmarks.benchmark

Quick Start

importasynciofromcacheflightimportCacheFlightflight=CacheFlight()
asyncdefget_user(user_id: str) ->dict:
returnawaitflight.run(
f"user:{user_id}",
lambda: fetch_user_from_db(user_id),
)
# 100 concurrent calls for the same user → 1 DB queryresults=awaitasyncio.gather(*(get_user("42") for_inrange(100)))

API

CacheFlight(options?)

Create an instance with optional configuration:

OptionTypeDefaultDescription
max_keysint | NoneNoneSafety cap on concurrent in-flight keys
default_timeout_msfloat | NoneNoneFallback timeout for all run() calls
on_eventCallable | NoneNoneCallback for observability events
fromcacheflightimportCacheFlight, CacheFlightOptionsflight=CacheFlight(CacheFlightOptions(
max_keys=10_000,
default_timeout_ms=5000,
on_event=lambdae: print(e.type, e.key),
))

await flight.run(key, task, options?)

Execute task for key, deduplicating concurrent calls.

  • key (str): Canonical identifier for the work.
  • task (Callable[[], Awaitable[T]]): Async callable that does the expensive work.
  • options (RunOptions): Per-call overrides (timeout_ms, metadata).

Cancellation semantics:

  • If a single caller cancels, other waiters still receive the result.
  • If all waiters cancel, the underlying task is cancelled and the entry is cleaned up.
fromcacheflightimportRunOptionsresult=awaitflight.run(
"report:daily",
generate_report,
RunOptions(timeout_ms=10_000),
)

flight.has(key) -> bool

Check if a key is currently in-flight.

flight.size -> int

Number of active in-flight keys.

flight.clear()

Cancel all in-flight entries. Waiters receive asyncio.CancelledError.

flight.cancel(key) -> bool

Cancel a specific in-flight key. Returns True if a key was cancelled, False if the key was not found.

importasynciofromcacheflightimportCacheFlight, CancelledErrorflight=CacheFlight()
asyncdefexample():
# Start a slow task in the backgroundtask=asyncio.create_task(
flight.run("report:daily", generate_report)
)
awaitasyncio.sleep(0.1) # let it start# Cancel it — all waiters will receive CancelledErrorcancelled=flight.cancel("report:daily")
print(f"Cancelled: {cancelled}") # Truetry:
awaittaskexceptCancelledError:
print("Task was cancelled as expected")

Events

Subscribe via on_event for observability:

Event TypeFieldsWhen
hitkeyCaller joined existing in-flight entry
misskeyNew task started
task_startedkeyTask execution began
task_resolvedkey, duration_msTask completed successfully
task_rejectedkey, duration_ms, errorTask raised an exception
task_timed_outkey, duration_msTask exceeded timeout
task_cancelledkey, duration_msTask was cancelled
max_keys_rejectedkeyKey rejected due to max_keys cap

Exceptions

ExceptionWhen
cacheflight.TimeoutErrorTask exceeds timeout_ms
cacheflight.MaxKeysErrorIn-flight registry at max_keys capacity
cacheflight.CancelledErrorUnderlying task cancelled (e.g., cancel() or all waiters cancel)

Key Design Guidelines

Good keys are deterministic and include all parameters that affect the result:

# Good: includes all result-shaping inputskey=f"user:{user_id}:locale:{locale}:role:{role}"# Bad: non-deterministic or missing contextkey=f"user:{user_id}"# locale/role changes give wrong cached result
  • Prefix with domain (user:, report:, search:)
  • Never put secrets or PII in plain-text keys
  • Hash long payloads to cap key size

Limitations

  • Single-process only — no cross-instance deduplication without a distributed coordination layer.
  • Not a cache — CacheFlight deduplicates in-flight execution; it does not store results.
  • Set timeouts for unreliable downstream dependencies.

Observability and Metrics (Suggested)

Typical event-to-metric mapping:

  • hit -> cacheflight_hit_total (counter)
  • miss -> cacheflight_miss_total (counter)
  • task_rejected -> cacheflight_error_total (counter)
  • task_timed_out -> cacheflight_timeout_total (counter)
  • task_cancelled -> cacheflight_cancelled_total (counter)
  • max_keys_rejected -> cacheflight_max_keys_rejected_total (counter)
  • task_resolved -> cacheflight_task_duration_ms (histogram)
  • flight.size -> cacheflight_active_keys (gauge)

Alerting ideas:

  • sustained task_timed_out ratio above baseline for 5+ minutes
  • max_keys_rejected_total > 0 (capacity pressure)
  • cacheflight_active_keys at or above 80% of max_keys for >10 minutes

Security and Redaction

Never log raw keys or metadata that can include secrets or PII. Hash or redact before emitting.

importhashlibfromcacheflightimportCacheFlight, CacheFlightOptionsdefon_event_safe(event):
key_hash=hashlib.sha256(event.key.encode()).hexdigest()[:12]
print(event.type, key_hash)
flight=CacheFlight(CacheFlightOptions(on_event=on_event_safe))

Benchmarks

Run the benchmark script and record environment details as described in benchmarks/README.md:

python -m benchmarks.benchmark

Anti-Patterns

  • Using non-canonical keys (e.g., json.dumps on unsorted dicts)
  • Treating CacheFlight as persistent cache storage
  • Omitting timeout for unreliable dependencies
  • Expecting cross-instance deduplication

License

MIT

About

Prevent duplicate expensive async work under concurrent load. When many callers request the same key simultaneously, CacheFlight ensures the task executes once and shares the result with all waiters.

Resources

Stars

0 stars

Watchers

0 watching

Forks

Releases

Packages

Contributors

Languages