Views

get_redis_value(key, redis_pool=Depends(get_redis_pool)) async

Get value from redis.

Parameters:

Name	Type	Description	Default
key	str	redis key, to get data from.	required
redis_pool	ConnectionPool	redis connection pool.	Depends(get_redis_pool)

Returns:

Type	Description
RedisValueDTO	information from redis.

Source code in hestia/web/api/redis/views.py

@router.get("/", response_model=RedisValueDTO)
async def get_redis_value(
    key: str,
    redis_pool: ConnectionPool = Depends(get_redis_pool),
) -> RedisValueDTO:
    """
    Get value from redis.

    :param key: redis key, to get data from.
    :param redis_pool: redis connection pool.
    :returns: information from redis.
    """
    async with Redis(connection_pool=redis_pool) as redis:  # type: Redis[Any]
        redis_value = await redis.get(key)
        await add.kiq(1, 2)

    return RedisValueDTO(
        key=key,
        value=redis_value,
    )

set_redis_value(redis_value, redis_pool=Depends(get_redis_pool)) async

Set value in redis.

Parameters:

Name	Type	Description	Default
redis_value	RedisValueDTO	new value data.	required
redis_pool	ConnectionPool	redis connection pool.	Depends(get_redis_pool)

Source code in hestia/web/api/redis/views.py

@router.put("/")
async def set_redis_value(
    redis_value: RedisValueDTO,
    redis_pool: ConnectionPool = Depends(get_redis_pool),
) -> None:
    """
    Set value in redis.

    :param redis_value: new value data.
    :param redis_pool: redis connection pool.
    """
    if redis_value.value is not None:
        async with Redis(connection_pool=redis_pool) as redis:  # type: Redis[Any]
            await redis.set(name=redis_value.key, value=redis_value.value)