Cashews

7.5.0 · active · verified Wed Apr 15

Cashews is a Python library providing asynchronous cache tools, designed to help build fast and reliable applications. It supports multiple storage backends like in-memory, Redis, and DiskCache, offering a decorator-based API and various caching strategies. The current version is 7.5.0, and it maintains an active release cadence with regular updates and community support.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates setting up Cashews with a Redis backend (or in-memory if Redis is unavailable) and using the `@cache` decorator for an asynchronous function. It shows how subsequent calls to the cached function with the same arguments will retrieve data from the cache rather than re-executing the heavy operation.

import asyncio
import os
from datetime import timedelta

from cashews import cache

# Configure Redis cache. Replace with your Redis URL or use "mem://" for in-memory.
# For production, use environment variables for connection strings.
REDIS_URL = os.environ.get("CASHEWS_REDIS_URL", "redis://localhost:6379/0")
cache.setup(REDIS_URL, client_side=True)

@cache(ttl=timedelta(minutes=5), key="user:{user_id}")
async def get_user_data(user_id: int):
    """
    Simulates a long-running operation to fetch user data.
    This function's result will be cached.
    """
    print(f"Fetching data for user_id: {user_id} from source...")
    await asyncio.sleep(1) # Simulate I/O delay
    return {"id": user_id, "name": f"User {user_id}", "email": f"user{user_id}@example.com"}

async def main():
    print("First call (should fetch from source):")
    user1_data = await get_user_data(1)
    print(f"Result: {user1_data}")

    print("\nSecond call (should hit cache):")
    user1_data_cached = await get_user_data(1)
    print(f"Result: {user1_data_cached}")

    print("\nFetching data for a different user (should fetch from source):")
    user2_data = await get_user_data(2)
    print(f"Result: {user2_data}")

    # Example of explicit cache invalidation:
    # await cache.invalidate(get_user_data, user_id=1)
    # print("\nAfter invalidation, first call again (should fetch from source):")
    # user1_data_recalled = await get_user_data(1)
    # print(f"Result: {user1_data_recalled}")

if __name__ == "__main__":
    asyncio.run(main())

view raw JSON →