coredis-utils
A collection of helpful utilities for coredis.
Features
- Caching decorator with thundering herd protection and error caching
- Idempotency keys
- Fixed-window rate limiting
Installation
$ pip install coredis-utils
Getting started
First, create a CoredisUtils object wrapping a coredis.Redis instance:
from coredis import Redis
from coredis_utils import CoredisUtils
client = Redis(...)
utils = CoredisUtils(client)
Caching is implemented with a decorator:
@utils.cached(ttl=60)
async def my_task() -> int: ...
Idempotency uses a simple check:
if await utils.idempotent("my-key", ttl=60):
... # code in this block can only run once
Rate limiting is similar:
from datetime import timedelta
for _ in range(10):
if await utils.limit("my-ip-addr", 5, timedelta(seconds=1)): # limit to 5/second
print("success")
success
success
success
success
success
Advanced caching
Cache keys are generated using a SHA256 hash of pickled arguments. You can exclude non-serializable arguments from cache key construction:
from sqlalchemy.ext.asyncio import AsyncSession
@utils.cached(ttl=60, exclude={"session"})
async def my_task(session: AsyncSession) -> int: ...
You can also customize which parts of arguments get hashed:
@utils.cached(
ttl=60,
key_fns={
# hash just the ID, not the entire model
"user": lambda u: u.id,
# hash a couple relevant fields
"message": lambda m: (m.type, m.timestamp),
},
)
async def my_task(user: User, message: Message) -> int: ...
Errors can be cached and propagated just like normal responses:
@utils.cached(ttl=60, error_ttl=5)
async def my_task() -> int:
raise Exception("Oh no!")
You can easily invalidate keys by passing the same arguments:
@utils.cached(ttl=60)
async def my_task(time: int) -> int: ...
await my_task.invalidate(3)
1 maintainer
Metadata
1 maintainer
Release files for coredis-utils 0.5.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| coredis_utils-0.5.0.tar.gz | 7.2 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| coredis_utils-0.5.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 15.4 kB
Release files / coredis_utils-0.5.0.tar.gz
| Download URL | coredis_utils-0.5.0.tar.gz |
|---|---|
| Size | 7.2 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
0c973ad4fd92c1563f1c30ef756de6581df229374985615c666c390776f349d9
|
|
BLAKE2b-256 checksum How to use checksums |
bfeffe8453bea8a089b9d769c0817cbe27ba04a94a16ede2cd2734b291630b12
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.11.32 {"installer":{"name":"uv","version":"0.11.32","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Release files / coredis_utils-0.5.0-py3-none-any.whl
| Download URL | coredis_utils-0.5.0-py3-none-any.whl |
|---|---|
| Size | 8.2 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
d1998268d5e358d7387feb582ee09d38751404bfcd3da1de53ae272a1aae8038
|
|
BLAKE2b-256 checksum How to use checksums |
c8d54f328e327e06310345557a4b67ee3314ad398e0c8879d702ebeae8aa2fb7
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.11.32 {"installer":{"name":"uv","version":"0.11.32","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
