Idempotency with coredis

idempotent.py

from datetime import timedelta
from coredis import Redis

redis = Redis.from_url("redis://localhost:6379", decode_responses=True)

async def idempotent(key: str, ttl: timedelta | int | None = 60) -> bool:
    """
    Shields code from being run multiple times.
    """
    return bool(
        await redis.set(f"pydempotent:{key}", 1, condition=PureToken.NX, ex=ttl)
    )
  
# Example usage
for _ in range(10):
    if await idempotent("test_key"):
        print("called!")  # this will only print once!