### Install Pre-commit Hooks

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Install pre-commit hooks for the development environment.

```shell
pip install pre-commit && pre-commit install --install-hooks
```

--------------------------------

### Install Cashews dependencies

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Install the core package and optional dependencies for specific backends or features.

```bash
pip install cashews
pip install cashews[redis]
pip install cashews[diskcache]
pip install cashews[dill] # can cache in redis more types of objects
pip install cashews[speedup] # for bloom filters
```

--------------------------------

### Configure Cache Backends

Source: https://context7.com/krukov/cashews/llms.txt

Setup various cache backends using connection strings, including Redis with client-side caching and prefix-based routing.

```python
import asyncio
from cashews import cache, Cache

# Simple in-memory setup
cache.setup("mem://")

# Redis setup with options
cache.setup(
    "redis://localhost:6379/0",
    db=1,
    secret="my_secret_key",
    digestmod="md5",
    client_side=True,  # Enable client-side caching for 10x performance
    suppress=True,     # Suppress connection errors
)

# Multiple backends with prefix routing
cache.setup("redis://localhost:6379/0")  # Default backend
cache.setup("mem://?size=500", prefix="user")  # Memory backend for user: keys

# DiskCache setup
cache.setup("disk://?directory=/tmp/cache&timeout=1&shards=12")

# Create custom cache instance
my_cache = Cache(name="custom")
my_cache.setup("mem://")

async def main():
    await cache.set("key", "value", expire="1h")
    result = await cache.get("key")
    print(result)  # Output: value
    await cache.close()

asyncio.run(main())
```

--------------------------------

### Run Tests with Pytest

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Install dependencies and run tests using pytest.

```shell
pip install .[tests,redis,diskcache,speedup] fastapi aiohttp requests httpx SQLAlchemy prometheus-client

pytest // run all tests with all backends
pytest -m "not redis" // all tests without tests for redis backend
```

--------------------------------

### Toggle cache state

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Enable or disable the cache during setup using the enable or disable parameters.

```python
cache.setup("redis://redis/0?enable=1")
cache.setup("mem://?size=500", disable=True)
cache.setup("mem://?size=500", enable=False)
```

--------------------------------

### Perform Basic Cache Operations

Source: https://context7.com/krukov/cashews/llms.txt

Execute common cache tasks including setting, getting, bulk operations, atomic increments, and key management.

```python
import asyncio
from decimal import Decimal
from cashews import cache

cache.setup("mem://")

async def basic_operations():
    # Set values with different TTL formats
    await cache.set("key1", {"name": "Alice", "age": 30}, expire="1h")
    await cache.set("key2", 42, expire=3600)  # 1 hour in seconds
    await cache.set("key3", "test", exist=None)  # Set only if not exists

    # Get values
    value = await cache.get("key1")  # Returns {"name": "Alice", "age": 30}
    value_with_default = await cache.get("missing", default="fallback")  # Returns "fallback"

    # Get or set (atomic operation)
    result = await cache.get_or_set("computed", default=lambda: "expensive_result", expire="10m")

    # Bulk operations
    await cache.set_many({"a": 1, "b": Decimal("10.5"), "c": [1, 2, 3]}, expire="1m")
    values = await cache.get_many("a", "b", "c")  # Returns (1, Decimal('10.5'), [1, 2, 3])

    # Increment
    count = await cache.incr("counter")  # Returns 1
    count = await cache.incr("counter", value=5)  # Returns 6

    # Check existence and TTL
    exists = await cache.exists("key1")  # True
    ttl = await cache.get_expire("key1")  # Seconds until expiration

    # Update TTL
    await cache.expire("key1", timeout="2h")

    # Delete operations
    await cache.delete("key1")
    await cache.delete_many("key2", "key3")
    await cache.delete_match("prefix:*")  # Delete by pattern

    # Scan keys
    async for key in cache.scan("user:*"):
        print(f"Found key: {key}")

    # Get keys with values
    async for key, value in cache.get_match("config:*"):
        print(f"{key} = {value}")

    # Utility methods
    total_keys = await cache.get_keys_count()
    await cache.ping()
    await cache.clear()

asyncio.run(basic_operations())
```

--------------------------------

### Get Cache Value by Key

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Retrieve the value associated with a given key. Returns `default` if the key is not found.

```python
await cache.get("key", default=None)  # -> Any
```

--------------------------------

### Get Multiple Cache Values

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Retrieve values for multiple keys. Returns a tuple of values in the order of the requested keys.

```python
await cache.get_many("key1", "key2", default=None)  # -> tuple[Any]
```

--------------------------------

### Get Total Key Count

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Retrieve the total number of keys currently stored in the cache.

```python
await cache.get_keys_count()  # -> int - total number of keys in cache
```

--------------------------------

### Use Template Context for Cache Keys

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Incorporate variables from a template context into cache keys using the `@{get(variable_name)}` syntax. The context must be provided during the function call.

```python
from cashews import cache, key_context

cache.setup("mem://")


@cache(ttl="2h", key="user:{@:get(client_id)}")
async def get_current_user():
  pass

...
with key_context(client_id=135356):
    await get_current_user()
```

--------------------------------

### Get Key Expiration Time

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Retrieve the remaining time in seconds until a key expires. Returns -1 if the key never expires or does not exist.

```python
await cache.get_expire("key")  # -> int seconds to expire
```

--------------------------------

### Identify Inconsistent Cache Operations

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Example of code where cache operations might become inconsistent during database transaction rollbacks.

```python
async def my_handler():
    async with db.transaction():
        await db.insert(user)
        await cache.set(f"key:{user.id}", user)
        await api.service.register(user)
```

```python
async def login(user, token, session):
    ...
    old_session = await cache.get(f"current_session:{user.id}")
    await cache.incr(f"sessions_count:{user.id}")
    await cache.set(f"current_session:{user.id}", session)
    await cache.set(f"token:{token.id}", user)
    return old_session
```

--------------------------------

### Get or Set Cache Value

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Retrieve a value by key, or if not found, compute and set it using the provided awaitable or callable, with an optional expiration time.

```python
await cache.get_or_set("key", default=awaitable_or_callable, expire="1h")  # -> Any
```

--------------------------------

### Configure cache backends

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Set up the cache using connection strings or keyword arguments.

```python
from cashews import cache

# via url
cache.setup("redis://0.0.0.0/?db=1&socket_connect_timeout=0.5&suppress=0&secret=my_secret&enable=1")
# or via kwargs
cache.setup("redis://0.0.0.0/", db=1, wait_for_connection_timeout=0.5, suppress=False, secret=b"my_key", enable=True)
```

--------------------------------

### Configure Simple Cache Strategy

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Sets up a basic cache with a specified Time-To-Live (TTL) and a dynamic key generation based on request user ID. Requires `timedelta` for TTL.

```python
from datetime import timedelta
from cashews import cache

cache.setup("mem://")

@cache(ttl=timedelta(hours=3), key="user:{request.user.uid}")
async def long_running_function(request):
    ...
```

--------------------------------

### Get Raw Cache Value

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Retrieve the raw, un-deserialized value associated with a key.

```python
await cache.get_raw("key") # -> Any
```

--------------------------------

### Configure DiskCache Backend

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Set up the DiskCache backend, optionally specifying directory, timeout, and number of shards. Disable shards by setting 'shards=0'.

```python
cache.setup("disk://")
```

```python
cache.setup("disk://?directory=/tmp/cache&timeout=1&shards=0")  # disable shards
```

```python
Gb = 1073741824
cache.setup("disk://", size_limit=3 * Gb, shards=12)
```

--------------------------------

### Implement basic caching

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Configure the cache backend and use either the decorator-based API or direct function calls.

```python
from cashews import cache

cache.setup("mem://")  # configure as in-memory cache, but redis/diskcache is also supported

# use a decorator-based API
@cache(ttl="3h", key="user:{request.user.uid}")
async def long_running_function(request):
    ...

# or for fine-grained control, use it directly in a function
async def cache_using_function(request):
    await cache.set(key=request.user.uid, value=request.user, expire="20h")
    ...
```

--------------------------------

### Configure Redis Cache Backend

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Set up the Redis backend with various options including database, password, timeouts, security, serialization, compression, and clustering.

```python
cache.setup("redis://0.0.0.0/?db=1&minsize=10&suppress=false&secret=my_secret", prefix="func")
```

```python
cache.setup("redis://0.0.0.0/2", password="my_pass", socket_connect_timeout=0.1, retry_on_timeout=True, secret="my_secret")
```

```python
cache.setup("redis://0.0.0.0", client_side=True, client_side_prefix="my_prefix:", pickle_type="dill", compress_type="gzip")
```

```python
cache.setup("redis://0.0.0.0:6379", cluster=True)
```

--------------------------------

### Initialize custom cache instance

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Create a standalone Cache instance instead of using the default global one.

```python
from cashews import Cache

cache = Cache()
cache.setup(...)
```

--------------------------------

### Configure multiple backends by prefix

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Route cache operations to different backends based on key prefixes.

```python
cache.setup("redis://redis/0")
cache.setup("mem://?size=500", prefix="user")

await cache.get("accounts")  # will use the redis backend
await cache.get("user:1")  # will use the memory backend
```

--------------------------------

### Configure Cache for FastAPI

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Set up the cache with the required prefix for FastAPI middleware.

```python
from cashews import cache
cache.setup(...)
# or
cache.setup(..., prefix="fastapi:")
```

--------------------------------

### Initialize Cache Transactions

Source: https://context7.com/krukov/cashews/llms.txt

Set up atomic cache operations with transaction support.

```python
import asyncio
from cashews import cache, TransactionMode

cache.setup("redis://localhost:6379")
```

--------------------------------

### Run Tests with Tox

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Execute test suites for different backends using tox.

```shell
pip install tox
tox -e py  // tests for inmemory backend
tox -e py-diskcache  // tests for diskcache backend
tox -e py-redis  // tests for redis backend  - you need to run redis
tox -e py-redis_cluster  // tests for redis cluster backend  - you need to run redis cluster
tox -e py-integration  // tests for integrations with aiohttp and fastapi

tox // to run all tests for all python that is installed on your machine
```

--------------------------------

### Configure In-Memory Cache

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Initialize the cache to use in-memory storage.

```python
cache.setup("mem://")  # configure as in-memory cache
```

--------------------------------

### Enable Prometheus Metrics

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Configure the Prometheus middleware to track cache metrics.

```python
from cashews import cache
from cashews.contrib.prometheus import create_metrics_middleware

metrics_middleware = create_metrics_middleware(with_tag=False)
cache.setup("redis://", middlewares=(metrics_middleware,))
```

--------------------------------

### Register Cache Callbacks

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Use CallbackMiddleware to execute functions before specific cache commands are triggered.

```python
from cashews import cache, Command


def callback(key, result):
  print(f"GET key={key}")

with cache.callback(callback, cmd=Command.GET):
    await cache.get("test")  # also will print "GET key=test"
```

--------------------------------

### Configure in-memory cache

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Set up the in-memory LRU cache with optional size and expiration check intervals.

```python
cache.setup("mem://")
cache.setup("mem://?check_interval=10&size=10000")
```

--------------------------------

### Integrate Cashews with FastAPI

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Apply caching middlewares and decorators to FastAPI routes.

```python
from fastapi import FastAPI, Header, Query
from fastapi.responses import StreamingResponse

from cashews import cache
from cashews.contrib.fastapi import (
    CacheDeleteMiddleware,
    CacheEtagMiddleware,
    CacheRequestControlMiddleware,
    cache_control_ttl,
)

app = FastAPI()
app.add_middleware(CacheDeleteMiddleware)
app.add_middleware(CacheEtagMiddleware)
app.add_middleware(CacheRequestControlMiddleware)
metrics_middleware = create_metrics_middleware()
cache.setup(os.environ.get("CACHE_URI", "redis://"))



@app.get("/")
@cache.failover(ttl="1h")
@cache(ttl=cache_control_ttl(default="4m"), key="simple:{user_agent:hash}", time_condition="1s")
async def simple(user_agent: str = Header("No")):
    ...


@app.get("/stream")
@cache(ttl="1m", key="stream:{file_path}")
async def stream(file_path: str = Query(__file__)):
    return StreamingResponse(_read_file(file_path=file_path))


async def _read_file(_read_file):
    ...
```

--------------------------------

### Configure TTL as String (Golang-like Format)

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Define the cache TTL using a string format similar to Golang's duration strings (e.g., '10m', '2h30s'). This provides a concise way to specify TTLs.

```python
@cache(ttl="10m")
async def get(item_id: int) -> Item:
    pass
```

--------------------------------

### Define Custom Cache Middleware

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Create custom middleware to log cache requests by defining an async function and registering it via cache.setup.

```python
import logging
from cashews import cache

logger = logging.getLogger(__name__)


async def logging_middleware(call, cmd: Command, backend: Backend, *args, **kwargs):
    key = args[0] if args else kwargs.get("key", kwargs.get("pattern", ""))
    logger.info("=> Cache request: %s ", cmd.value, extra={"args": args, "cache_key": key})
    return await call(*args, **kwargs)


cache.setup("mem://", middlewares=(logging_middleware, ))
```

--------------------------------

### Implement Cache Middleware and Detection

Source: https://context7.com/krukov/cashews/llms.txt

Use custom middleware for logging or metrics and the cache.detect context manager to track cache hits and misses.

```python
import asyncio
import logging
from cashews import cache, Command
from cashews.backends.interface import Backend

cache.setup("mem://")
logger = logging.getLogger(__name__)

# Logging middleware
async def logging_middleware(call, cmd: Command, backend: Backend, *args, **kwargs):
    key = args[0] if args else kwargs.get("key", kwargs.get("pattern", ""))
    logger.info(f"Cache {cmd.value}: {key}")
    result = await call(*args, **kwargs)
    logger.info(f"Cache {cmd.value} completed: {key}")
    return result

# Setup with middleware
cache.setup("mem://", middlewares=(logging_middleware,))

@cache(ttl="5m")
async def cached_function(param: str):
    return f"result:{param}"

async def detect_cache_usage():
    # Detect cache hits/misses
    with cache.detect as detector:
        result = await cached_function("test")
        calls = detector.calls

        for key, call_info in calls.items():
            print(f"Key: {key}")
            for info in call_info:
                print(f"  TTL: {info.get('ttl')}, Strategy: {info.get('name')}")

    # Callback on cache operations
    def on_get(key, result):
        print(f"GET {key}: {'HIT' if result else 'MISS'}")

    with cache.callback(on_get, cmd=Command.GET):
        await cache.get("test_key")

asyncio.run(detect_cache_usage())
```

--------------------------------

### Configure Redis with SSL

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Use 'rediss' schema for secure connections to Redis, specifying SSL certificate paths.

```python
cache.setup("rediss://0.0.0.0/", ssl_ca_certs="path/to/ca.crt", ssl_keyfile="path/to/client.key",ssl_certfile="path/to/client.crt",)
```

--------------------------------

### FastAPI Integration with Middlewares

Source: https://context7.com/krukov/cashews/llms.txt

Integrate Cashews with FastAPI using middleware for ETag support, Cache-Control headers, and cache deletion. The order of middleware addition is important.

```python
import asyncio
from fastapi import FastAPI, Header
from cashews import cache
from cashews.contrib.fastapi import (
    CacheDeleteMiddleware,
    CacheEtagMiddleware,
    CacheRequestControlMiddleware,
    cache_control_ttl,
)

app = FastAPI()

# Add middlewares (order matters)
app.add_middleware(CacheDeleteMiddleware)
app.add_middleware(CacheEtagMiddleware)
app.add_middleware(CacheRequestControlMiddleware)

# Configure cache
cache.setup("redis://localhost:6379")

# Dynamic TTL from Cache-Control header
@app.get("/data")
@cache(ttl=cache_control_ttl(default="5m"), key="data:{user_agent:hash}")
async def get_data(user_agent: str = Header("unknown")):
    await asyncio.sleep(1)
    return {"data": "response", "user_agent": user_agent}

# Standard caching with tags
@app.get("/users/{user_id}")
@cache(ttl="10m", key="user:{user_id}", tags=["users"])
async def get_user(user_id: int):
    return {"user_id": user_id, "name": "John"}

# Invalidate cache endpoint
@app.post("/users/{user_id}")
@cache.invalidate("user:{user_id}")
async def update_user(user_id: int):
    return {"updated": True}

# Client requests with headers:
# - Cache-Control: no-cache (bypass cache read)
# - Cache-Control: no-store (bypass cache read/write)
# - Cache-Control: max-age=300 (override TTL)
# - If-None-Match: "etag" (return 304 if unchanged)
# - Clear-Site-Data: cache (force cache miss)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
```

--------------------------------

### Iterate Over Matching Keys and Values

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Asynchronously iterate over keys and their corresponding values that match a given pattern. `batch_size` controls the number of keys fetched per iteration.

```python
async for key, value in cache.get_match("pattern:*", batch_size=100):
    ...
```

--------------------------------

### Disable and Enable Cache Commands

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Demonstrates how to disable or enable specific cache commands or all cache operations at runtime. Use `Command` enum for specific commands.

```python
from cashews import cache, Command

cache.setup("mem://")  # configure as in-memory cache

cache.disable(Command.DELETE)
cache.disable()
cache.enable(Command.GET, Command.SET)
cache.enable()

with cache.disabling():
  ...
```

--------------------------------

### Define Custom Cache Key Manually

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Manually define a cache key using string formatting for advanced scenarios. Ensure the cache is set up before defining decorated functions.

```python
from cashews import cache

cache.setup("mem://")

@cache(ttl="2h", key="user_info:{user_id}")
async def get_info(user_id: str):
    ...
```

--------------------------------

### Set Multiple Cache Key-Value Pairs

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Set multiple keys with their corresponding values simultaneously.

```python
await cache.set_many({"key1": value, "key2": value})  # -> None
```

--------------------------------

### Configure Hit Cache Strategy

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Sets up a cache that expires after a certain number of calls (`cache_hits`) and can update the cache after a specified duration (`update_after`).

```python
from cashews import cache

cache.setup("mem://")

@cache.hit(ttl="2h", cache_hits=100, update_after=2)
async def get(name):
    value = await api_call()
    return {"status": value}
```

--------------------------------

### Use Built-in Functions for Cache Key Formatting

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Employ built-in functions like `lower`, `upper`, `len`, `jwt`, and `hash` within cache key templates for value transformation. Specify hashing algorithms like `sha1`.

```python
@cache(ttl="2h", key="user_info:{user.name:lower}:{password:hash(sha1)}")
async def get_info(user: User, password: str):
    ...
```

```python
@cache(ttl="2h", key="user:{token:jwt(client_id)}")
async def get_user_by_token(token: str) -> User:
    ...
```

--------------------------------

### Scan Cache Keys

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Asynchronously iterate over all keys in the cache that match a given pattern. Note: This may not work with sharded DiskCache.

```python
async for key in cache.scan("pattern:*"):
    ...
```

--------------------------------

### Register Custom Key Formatters

Source: https://context7.com/krukov/cashews/llms.txt

Define custom transformations and type-specific formatters for key templates using the default_formatter registry.

```python
from cashews import cache, default_formatter
from decimal import Decimal

cache.setup("mem://")

# Register custom transformation
@default_formatter.register("prefix")
def _prefix(value, chars=3):
    return value[:chars].upper()

# Register type formatter
@default_formatter.type_format(Decimal)
def _decimal(value: Decimal) -> str:
    return str(value.quantize(Decimal("0.00")))

# Use custom formatters in key templates
@cache(ttl="1h", key="user:{username:prefix(4)}:data")
async def get_user_data(username: str):
    return {"username": username}

@cache(ttl="30m", key="price:{amount}:{currency:upper}")
async def convert_price(amount: Decimal, currency: str):
    return {"amount": amount, "currency": currency}

# Built-in formatters: lower, upper, len, jwt, hash
@cache(ttl="1h", key="session:{token:jwt(user_id)}")
async def get_session(token: str):
    return {"active": True}

@cache(ttl="1h", key="auth:{password:hash(sha256)}")
async def authenticate(password: str):
    return {"authenticated": True}
```

--------------------------------

### Cache with Lock and Tags

Source: https://context7.com/krukov/cashews/llms.txt

Use `lock=True` to prevent cache stampedes for expensive computations. Employ `tags` for efficient cache invalidation.

```python
import asyncio
from cashews import cache

@cache(ttl="5m", key="expensive:{param}", lock=True)
async def expensive_computation(param: str):
    await asyncio.sleep(2)
    return f"computed:{param}"

@cache(ttl="1h", tags=["users", "user:{user_id}"])
async def get_user_data(user_id: int):
    return {"id": user_id, "data": "..."}

async def main():
    # First call - executes function
    user = await get_user(1)  # Takes 0.5s
    print(user)  # {"id": 1, "name": "User 1"}

    # Second call - returns from cache
    user = await get_user(1)  # Instant

    # Invalidate by tags
    await cache.delete_tags("user:1")

asyncio.run(main())
```

--------------------------------

### Use Objects and Attributes in Cache Keys

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Utilize objects within cache keys by accessing their attributes through templates. This allows for dynamic key generation based on object properties.

```python
@cache(ttl="2h", key="user_info:{user.uuid}")
async def get_info(user: User):
    ...
```

--------------------------------

### Configure Soft Cache Strategy

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Provides a cache with fail protection based on a soft TTL. If recalculation fails after the soft TTL, the current cached value is returned if it's not older than the main TTL.

```python
from cashews import cache

cache.setup("mem://")

# if you call this function after 7 min, cache will be updated and return a new result.
# If it fail on recalculation will return current cached value (if it is not more than 10 min old)
@cache.soft(ttl="10m", soft_ttl="7m")
async def get(name):
    value = await api_call()
    return {"status": value}
```

--------------------------------

### Check if Key Exists

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Check if a key exists in the cache.

```python
await cache.exists("key") # -> bool
```

--------------------------------

### Configure Early Cache Strategy

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Addresses cache stampede by recalculating a hot cache result in the background. The cache is updated in the background after a specified `early_ttl` and served until the main `ttl` expires.

```python
from cashews import cache  # or: from cashews import early

# if you call this function after 7 min, cache will be updated in a background
@cache.early(ttl="10m", early_ttl="7m")
async def get(name):
    value = await api_call()
    return {"status": value}
```

--------------------------------

### Cache Class Methods Without Self in Key

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Demonstrates how to cache class methods and avoid including the instance (`self`) in the cache key by default. The default key includes module, class, method, and arguments.

```python
from cashews import cache

cache.setup("mem://")

class MyClass:

    @cache(ttl="2h")
    async def get_name(self, user, version="v1"):
         ...

# a key template will be "__module__:MyClass.get_name:self:{self}:user:{user}:version:{version}"

await MyClass().get_name("me", version="v2")
# a key will be "__module__:MyClass.get_name:self:<__module__.MyClass object at 0x105edd6a0>:user:me:version:v1"
```

--------------------------------

### Customize Cache Key Templates

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Cashews automatically generates keys based on function metadata and arguments. You can observe how parameters are serialized into the cache key.

```python
from cashews import cache

cache.setup("mem://")

@cache(ttl=timedelta(hours=3))
async def get_name(user, *args, version="v1", **kwargs):
    ...

# a key template will be "__module__.get_name:user:{user}:{__args__}:version:{version}:{__kwargs__}"

await get_name("me", version="v2")
# a key will be "__module__.get_name:user:me::version:v2"
await get_name("me", version="v1", foo="bar")
# a key will be "__module__.get_name:user:me::version:v1:foo:bar"
await get_name("me", "opt", "attr", opt="opt", attr="attr")
```

--------------------------------

### Implement HTTP Middleware for Caching

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Use this middleware to inject cache-related headers into HTTP responses based on detected cache activity.

```python
@app.middleware("http")
async def add_from_cache_headers(request: Request, call_next):
    with cache.detect as detector:
        response = await call_next(request)
        if detector.calls:
            key = list(detector.calls.keys())[0]
            response.headers["X-From-Cache"] = key
            expire = await cache.get_expire(key)
            response.headers["X-From-Cache-Expire-In-Seconds"] = str(expire)
    return response
```

--------------------------------

### Configure TTL as Integer (Seconds)

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Set the cache time-to-live (TTL) using an integer representing the number of seconds. This is a straightforward way to define cache duration.

```python
from cashews import cache
from datetime import timedelta

cache.setup("mem://")

@cache(ttl=60 * 10)
async def get(item_id: int) -> Item:
    pass
```

--------------------------------

### Configure Failover Cache Strategy

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Implements a failover cache that returns cached results if specific exceptions are raised during the function execution. Exceptions can be specified, or a default can be set globally.

```python
from cashews import cache

cache.setup("mem://")

# note: the key will be "__module__.get_status:name:{name}"
@cache.failover(ttl="2h", exceptions=(ValueError, MyException))
async def get_status(name):
    value = await api_call()
    return {"status": value}
```

```python
cache.set_default_fail_exceptions(ValueError, MyException)
```

--------------------------------

### Register and Use Type-Specific Cache Key Formatters

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Register formatters for specific data types, such as `Decimal`, to control how they are represented in cache keys. Ensure the formatter returns a string.

```python
from decimal import Decimal
from cashews import default_formatter, cache

@default_formatter.type_format(Decimal)
def _decimal(value: Decimal) -> str:
    return str(value.quantize(Decimal("0.00")))


@cache(ttl="2h", key="price-{item.price}:{item.currency:upper}")  # a key will be "price-10.00:USD"
async def convert_price(item):
    ...
```

--------------------------------

### Configure TTL as timedelta Object

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Specify the cache TTL using a `timedelta` object for more expressive time durations. This is useful for defining TTLs in minutes, hours, etc.

```python
@cache(ttl=timedelta(minutes=10))
async def get(item_id: int) -> Item:
    pass
```

--------------------------------

### Set Cache Key-Value Pair

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Set a key with a value, optionally specifying expiration time and existence condition. `exist=None` overwrites if key exists.

```python
await cache.set(key="key", value=90, expire="2h", exist=None)  # -> bool
```

--------------------------------

### Register and Use Custom Cache Key Formatters

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Define and register custom formatter functions to transform values used in cache keys. The formatter can accept additional arguments.

```python
from cashews import default_formatter, cache

cache.setup("mem://")

@default_formatter.register("prefix")
def _prefix(value, chars=3):
    return value[:chars].upper()


@cache(ttl="2h", key="servers-user:{user.index:prefix(4)}")  # a key will be "servers-user:DWQS"
async def get_user_servers(user):
    ...
```

--------------------------------

### Configure Conditional Caching

Source: https://context7.com/krukov/cashews/llms.txt

Control caching behavior using result filters, exception handling, or execution time thresholds.

```python
import asyncio
from cashews import cache, NOT_NONE, with_exceptions, only_exceptions

cache.setup("mem://")

# Only cache non-None results
@cache(ttl="10m", condition=NOT_NONE)
async def search_user(username: str):
    if username == "admin":
        return {"username": "admin", "role": "admin"}
    return None  # Not cached

# Custom condition function
def cache_if_success(result, args, kwargs, key=None) -> bool:
    return result is not None and result.get("status") == "success"

@cache(ttl="1h", condition=cache_if_success)
async def process_request(request_id: int):
    return {"request_id": request_id, "status": "success"}

# Cache exceptions
@cache(ttl="5m", condition=with_exceptions(ValueError, TimeoutError))
async def may_fail(param: int):
    if param < 0:
        raise ValueError("Negative value")
    return param * 2

# Only cache exceptions (not successful results)
@cache(ttl="1m", condition=only_exceptions(ConnectionError))
async def external_call():
    raise ConnectionError("Service down")

# Time-based condition: only cache if execution > 1 second
@cache(ttl="30m", time_condition="1s")
async def expensive_query(query: str):
    import time
    time.sleep(1.5)  # Slow query - will be cached
    return {"query": query, "results": []}

async def main():
    # NOT_NONE condition
    result = await search_user("admin")  # Cached
    result = await search_user("unknown")  # Not cached (returns None)

    # Time condition
    result = await expensive_query("SELECT *")  # Cached (took > 1s)

asyncio.run(main())
```

--------------------------------

### Set Key Expiration Time

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Update the expiration time for an existing key.

```python
await cache.expire("key", timeout=10)
```

--------------------------------

### Implement Rate Limiting Decorators

Source: https://context7.com/krukov/cashews/llms.txt

Protects functions using fixed or sliding window rate limiting, with optional failover support.

```python
import asyncio
from cashews import cache, RateLimitError

cache.setup("mem://")

# Fixed window: max 10 calls per minute, ban for 5 minutes if exceeded
@cache.rate_limit(limit=10, period="1m", ttl="5m")
async def api_endpoint(user_id: int):
    return {"user_id": user_id, "data": "response"}

# Sliding window rate limit (smoother)
@cache.slice_rate_limit(limit=100, period="10m", key="user:{user_id}")
async def premium_endpoint(user_id: int):
    return {"premium": True}

# Combine with failover for graceful degradation
@cache.failover(ttl="10m", exceptions=(RateLimitError,))
@cache.rate_limit(limit=5, period="1m", ttl="10m")
async def protected_api(resource_id: int):
    return {"resource_id": resource_id}

async def main():
    # Normal calls succeed
    for i in range(10):
        try:
            result = await api_endpoint(1)
            print(f"Call {i+1}: Success")
        except RateLimitError:
            print(f"Call {i+1}: Rate limited!")

    # With failover, returns cached result when rate limited
    for i in range(20):
        result = await protected_api(1)
        print(f"Protected call {i+1}: {result}")

asyncio.run(main())
```

--------------------------------

### Cache Class Methods with Custom Key Template

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Customize the cache key for class methods by explicitly defining a `key` template. This allows control over which instance attributes or arguments are included.

```python
class MyClass:

    @cache(ttl="2h", key="{self._host}:name:{user}:{version}")
    async def get_name(self, user, version="v1"):
         ...

await MyClass(host="http://example.com").get_name("me", version="v2")
# a key will be "http://example.com:name:me:v1"
```

--------------------------------

### Detect Cache Source with Context Manager

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Use the `cache.detect` context manager to determine if a result was retrieved from the cache or obtained via a direct function call. It provides a `calls` attribute to inspect cache interactions.

```python
from cashews import cache

with cache.detect as detector:
    response = await something_that_use_cache()
    calls = detector.calls

print(calls)
```

--------------------------------

### Implement Bloom Filters for Membership Testing

Source: https://context7.com/krukov/cashews/llms.txt

Use the @cache.bloom decorator to perform space-efficient membership checks. The decorated function acts as a fallback for cache misses or potential false positives.

```python
import asyncio
from cashews import cache

cache.setup("redis://localhost:6379")

@cache.bloom(capacity=10_000, false_positives=1, name="emails")
async def email_exists(email: str) -> bool:
    # This function body is called only for cache misses
    # and when bloom filter indicates possible membership
    return await db_check_email(email)

async def main():
    # Add known emails to bloom filter
    for email in ["user1@example.com", "user2@example.com"]:
        await email_exists.set(email)

    # Check membership (fast, may have false positives)
    exists = await email_exists("user1@example.com")  # True
    exists = await email_exists("unknown@example.com")  # Likely False

asyncio.run(main())
```

--------------------------------

### Ping Cache Server

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Send a PING command to the cache server to check connectivity. Returns the server's response.

```python
await cache.ping(message=None)  # -> bytes
```

--------------------------------

### Transaction as Context Manager

Source: https://context7.com/krukov/cashews/llms.txt

Utilize async with cache.transaction() for managing cache operations within a block. The transaction commits on successful exit and rolls back on exceptions. Explicit rollback is also supported.

```python
import asyncio
from cashews import cache
from cashews.transaction import TransactionMode

async def transaction_example():
    # Context manager transaction
    async with cache.transaction() as tx:
        await cache.set("key1", "value1")
        await cache.set("key2", "value2")
        await cache.incr("counter")
        # Commits on exit, rollback on exception

    # With explicit rollback
    async with cache.transaction() as tx:
        await cache.set("temp", "data")
        if some_condition:
            await tx.rollback()
            return
        # Otherwise commits

    # Different isolation modes
    # FAST: Memory-based, no race protection (0-7% overhead)
    async with cache.transaction(TransactionMode.FAST):
        await cache.set("fast_key", "value")

    # LOCKED: Per-key locking (4-9% overhead, default)
    async with cache.transaction(TransactionMode.LOCKED, timeout=5):
        await cache.set("locked_key", "value")

    # SERIALIZABLE: Global lock, one transaction at a time
    async with cache.transaction(TransactionMode.SERIALIZABLE):
        await cache.set("serial_key", "value")

asyncio.run(transaction_example())
```

--------------------------------

### Configure Latency-Based Caching

Source: https://github.com/krukov/cashews/blob/master/Readme.md

The time_condition parameter allows caching only if the function execution exceeds a specified minimum latency.

```python
from cashews import cache

cache.setup("mem://")

@cache(ttl="1h", time_condition="3s")  # to cache for 1 hour if execution takes more than 3 seconds
async def get():
    ...
```

--------------------------------

### Define __str__ for Class Instance in Cache Key

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Improve class method caching by defining a `__str__` method. This method's return value will be used in the cache key instead of the default object representation.

```python
class MyClass:

    @cache(ttl="2h")
    async def get_name(self, user, version="v1"):
         ...

    def __str__(self) -> str:
        return self._host

await MyClass(host="http://example.com").get_name("me", version="v2")
# a key will be "__module__:MyClass.get_name:self:http://example.com:user:me:version:v1"
```

--------------------------------

### Clear Entire Cache

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Remove all keys and values from the cache.

```python
await cache.clear()
```

--------------------------------

### Configure Background Cache Refresh

Source: https://context7.com/krukov/cashews/llms.txt

Uses the hit decorator to trigger background cache refreshes after a specified number of hits.

```python
# Refresh in background after 800 hits, expire at 1000
@cache.hit(ttl="1h", cache_hits=1000, update_after=800)
async def get_popular_item(item_id: int):
    return {"item_id": item_id, "views": 5000}

async def main():
    # Each call decrements hit counter
    for i in range(10):
        product = await get_product(1)
        print(f"Call {i+1}: {product}")

    # After 1000 calls, cache refreshes

asyncio.run(main())
```

--------------------------------

### Prometheus Metrics Integration

Source: https://context7.com/krukov/cashews/llms.txt

Export cache metrics to Prometheus for monitoring. This involves creating a metrics middleware and mounting a Prometheus metrics endpoint.

```python
from cashews import cache
from cashews.contrib.prometheus import create_metrics_middleware
from prometheus_client import make_asgi_app
from fastapi import FastAPI

app = FastAPI()

# Create metrics middleware
metrics_middleware = create_metrics_middleware(
    with_tag=True  # Include cache tags in metrics
)

# Setup cache with metrics
cache.setup(
    "redis://localhost:6379",
    middlewares=(metrics_middleware,)
)

# Mount Prometheus metrics endpoint
metrics_app = make_asgi_app()
app.mount("/metrics", metrics_app)

@app.get("/")
@cache(ttl="5m", tags=["api"])
async def root():
    return {"message": "Hello"}

# Metrics exposed at /metrics include:
# - cashews_operations_total{operation="get|set|delete", status="hit|miss"}
# - cashews_operation_duration_seconds{operation="get|set|delete"}
# - cashews_cache_size_bytes
```

--------------------------------

### Implement Circuit Breaker Pattern

Source: https://context7.com/krukov/cashews/llms.txt

Opens a circuit when error rates exceed thresholds to prevent cascading failures.

```python
import asyncio
from cashews import cache, CircuitBreakerOpen

cache.setup("mem://")

# Open circuit when 10% of calls fail within 10 minutes
# Stay open for 5 minutes, then half-open for 1 minute
@cache.circuit_breaker(
    errors_rate=10,      # 10% error threshold
    period="10m",        # Measurement window
    ttl="5m",            # Time circuit stays open
    half_open_ttl="1m",  # Half-open period for testing
    min_calls=4          # Minimum calls before circuit can open
)
async def unreliable_service(param: str):
    import random
    if random.random() < 0.2:
        raise Exception("Service error")
    return f"result:{param}"

# Combine with failover for resilience
@cache.failover(ttl="1h", exceptions=(CircuitBreakerOpen,))
@cache.circuit_breaker(errors_rate=20, period="5m", ttl="2m")
async def external_api_call(endpoint: str):
    return {"endpoint": endpoint, "status": "ok"}

async def main():
    for i in range(30):
        try:
            result = await unreliable_service("test")
            print(f"Call {i+1}: {result}")
        except CircuitBreakerOpen:
            print(f"Call {i+1}: Circuit open, service unavailable")
        except Exception as e:
            print(f"Call {i+1}: Error - {e}")
        await asyncio.sleep(0.1)

asyncio.run(main())
```

--------------------------------

### Set Raw Cache Value

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Set a key with a raw string value without serialization.

```python
await cache.set_raw(key="key", value="str")  # -> bool
```

--------------------------------

### Configure TTL as a Callable Function

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Dynamically determine the cache TTL by providing a callable function. This function receives the decorated function's arguments and returns the TTL value.

```python
def _ttl(item_id: int) -> str:
    return "2h" if item_id > 10 else "1h"

@cache(ttl=_ttl)
async def get(item_id: int) -> Item:
    pass
```

--------------------------------

### Implement Distributed Locking

Source: https://context7.com/krukov/cashews/llms.txt

Coordinate access to shared resources across multiple processes using distributed locks.

```python
import asyncio
from cashews import cache

cache.setup("redis://localhost:6379")

async def distributed_lock_example():
    # Context manager lock
    async with cache.lock("resource:123", expire="30s"):
        # Exclusive access to resource
        print("Processing resource 123")
        await asyncio.sleep(2)

    # Check if locked
    is_locked = await cache.is_locked("resource:123", wait=0)
    print(f"Is locked: {is_locked}")

    # Wait for lock with timeout
    is_available = await cache.is_locked("resource:123", wait=10, step=0.5)

    # Manual lock management
    lock_acquired = await cache.set_lock("job:456", value="worker-1", expire="1m")
    if lock_acquired:
        try:
            # Do work
            pass
        finally:
            await cache.unlock("job:456", "worker-1")

asyncio.run(distributed_lock_example())
```

--------------------------------

### Acquire Cache Lock

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Acquire a distributed lock for a key with a specified expiration time. The lock is automatically released when exiting the `async with` block.

```python
async with cache.lock("key", expire=10):
    ...
```

--------------------------------

### Cache Invalidation with Tags

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Utilize tags for efficient cache invalidation, especially with Redis, by storing tagged keys in separate sets. Ensure tags backend is set up.

```python
from cashews import cache

cache.setup("redis://", client_side=True)

@cache(ttl="1h", tags=["items", "page:{page}"])
async def items(page=1):
    ...
```

```python
await cache.delete_tags("page:1")
await cache.delete_tags("items")
```

```python
cache.register_tag("my_tag", key_template="key{i}")

await cache.set("key1", "value", expire="1d", tags=["my_tag"])
```

--------------------------------

### Close Cache Connection

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Close the connection to the cache backend.

```python
await cache.close()
```

--------------------------------

### Configure Rate Limit Cache Strategy

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Applies rate limiting to function calls, raising a `RateLimitError` if the limit is reached. This decorator does not cache results; it can be combined with failover decorators.

```python
from cashews import cache, RateLimitError

cache.setup("mem://")

# no more than 10 calls per minute or ban for 10 minutes - raise RateLimitError
@cache.rate_limit(limit=10, period="1m", ttl="10m")
async def get(name):
    value = await api_call()
    return {"status": value}
```

```python
# no more than 100 calls in 10 minute window. if rate limit will rich -> return from cache
@cache.failover(ttl="10m", exceptions=(RateLimitError, ))
@cache.slice_rate_limit(limit=100, period="10m")
async def get_next(name):
    value = await api_call()
    return {"status": value}
```

--------------------------------

### Delete Keys Matching Pattern

Source: https://github.com/krukov/cashews/blob/master/Readme.md

Remove all keys from the cache that match a given pattern.

```python
await cache.delete_match("pattern:*")
```