Basically the RedisStore class has no method for batch operations, while if i have a redis instance i can use redis.pipeline(), the store class also makes the redis instance a private variable so i shouldn't be accessing that as well so, what should i do in this case? should i save the redis instance i pass to the store in app.state as well? and access it from there?
#redis store doesnt allow for batch operations
fast sable
also if i do that then i loose the namespacing benifit the store does automatically for me
fast sable
for now i did this,
class ActiveAccessTokenService:
"""ActiveAccessTokenService."""
__slots__ = ("_redis", "_conn", "_make_key")
_redis: Redis[bytes]
_conn: Connection
def __init__(self, request: Request[Any, Any, Any], db_connection: Connection) -> None:
store: RedisStore = request.app.stores.get(APP_CONFIG.access_token.blacklist_store) # pyright: ignore[reportAssignmentType]
self._redis = store._redis # pyright: ignore[reportPrivateUsage, reportUnknownMemberType]
self._make_key = store._make_key # pyright: ignore[reportPrivateUsage]
self._conn = db_connection
...
async def blacklist_active_tokens(self, *, user_id: int) -> None:
"""Blacklist all active tokens of a user in redis."""
tokens = await self.fetch_active_tokens(user_id=user_id, fields=("jti", "expires_at"))
pipe = self._redis.pipeline(transaction=False)
for token in tokens:
pipe.set(token.jti, "", max(1, token.expires_in))
await pipe.execute() # pyright: ignore[reportUnknownMemberType]
but this obviously uses the internals so
vocal veldt
Why don't you just extend the existing store with whatever methods you want?
and use it?
fast sable
but doesnt it have the same problem of the private variable usage having no garuantee?