| Server IP : 185.124.137.13 / Your IP : 216.73.217.1 Web Server : LiteSpeed System : Linux id-dci-web1986.main-hosting.eu 5.14.0-611.26.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jan 29 05:24:47 EST 2026 x86_64 User : u686484674 ( 686484674) PHP Version : 8.0.30 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/opt/alt/python311/lib/python3.11/site-packages/redis/ |
Upload File : |
from contextlib import contextmanager
try:
import hiredis # noqa
HIREDIS_AVAILABLE = True
except ImportError:
HIREDIS_AVAILABLE = False
try:
import cryptography # noqa
CRYPTOGRAPHY_AVAILABLE = True
except ImportError:
CRYPTOGRAPHY_AVAILABLE = False
def from_url(url, **kwargs):
"""
Returns an active Redis client generated from the given database URL.
Will attempt to extract the database id from the path url fragment, if
none is provided.
"""
from redis.client import Redis
return Redis.from_url(url, **kwargs)
@contextmanager
def pipeline(redis_obj):
p = redis_obj.pipeline()
yield p
p.execute()
def str_if_bytes(value):
return (
value.decode("utf-8", errors="replace") if isinstance(value, bytes) else value
)
def safe_str(value):
return str(str_if_bytes(value))
def dict_merge(*dicts):
"""
Merge all provided dicts into 1 dict.
*dicts : `dict`
dictionaries to merge
"""
merged = {}
for d in dicts:
merged.update(d)
return merged
def list_keys_to_dict(key_list, callback):
return dict.fromkeys(key_list, callback)
def merge_result(command, res):
"""
Merge all items in `res` into a list.
This command is used when sending a command to multiple nodes
and they result from each node should be merged into a single list.
res : 'dict'
"""
result = set()
for v in res.values():
for value in v:
result.add(value)
return list(result)