blob: 7cc8e10d0319fcf3ddf156ce405489b9748c0e6a (
about) (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
"""Connections for Redis."""
import logging
import traceback
import contextlib
from typing import Iterator
from redis import Redis, RedisError, ConnectionError as RedisConnectionError
@contextlib.contextmanager
def connection(redis_uri) -> Iterator[Redis]:
"""Connection to redis"""
rconn = Redis.from_url(redis_uri, decode_responses=True)
try:
if not rconn.ping():
raise RedisConnectionError("Could not connect to Redis.")
yield rconn
except RedisError as _rerr:
logging.debug(traceback.format_exc())
raise
finally:
rconn.close()
|
