Redis в Python: библиотека redis-py
Как работать с Redis в Python через redis-py: установка сервера и библиотеки, подключение, строки, хэши, списки, TTL, pipeline, pub/sub и redis.asyncio.
Содержание · 18 разделов
- Что такое Redis и для чего он нужен
- Redis или Valkey: какой сервер выбрать
- Как установить сервер Redis
- Docker: любая ОС
- Windows: WSL или Docker
- macOS и Linux
- Установка redis-py
- Подключение к Redis
- Строка подключения: redis:// и rediss://
- decode_responses: байты или строки
- Пул соединений и потоки
- Таймауты и повторы в redis-py 8
- Строки и счётчики
- Какие значения можно записать
- Время жизни ключей: TTL
- Хэши: объект с полями
- Списки: очередь задач
- Множества и отсортированные множества
- Какую структуру выбрать
- Поиск ключей: scan_iter вместо keys
- Pipeline: несколько команд за один запрос
- Транзакции и WATCH
- Блокировка lock
- Pub/Sub: публикация и подписка
- Асинхронный клиент redis.asyncio
- Частые ошибки
- Частые вопросы
- Как сохранить словарь или объект в Redis?
- Как проверить код без запущенного Redis?
- Чем redis.Redis отличается от redis.StrictRedis?
- Сколько баз данных в Redis и стоит ли их использовать?
- Как использовать Redis в Django?
- Итог
Для работы с Redis в Python используют библиотеку redis-py: она ставится командой pip install redis, а каждая команда Redis вызывается одноимённым методом клиента — SET превращается в r.set(), HGETALL в r.hgetall(). Минимальный пример:
import redis
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
r.set("greeting", "Привет, Redis", ex=60) # ключ проживёт 60 секунд
print(r.get("greeting")) # → Привет, Redis
Ниже — какой сервер выбрать (Redis или Valkey) и как запустить его на Windows, macOS и Linux, подключение и параметр decode_responses, строки, хэши, списки, множества, время жизни ключей, pipeline и транзакции, pub/sub, асинхронный клиент redis.asyncio и частые ошибки. Примеры проверены на redis-py 8.1 и Python 3.12.
Что такое Redis и для чего он нужен
Redis — сервер «ключ — значение», который держит данные в оперативной памяти. Чтение и запись занимают доли миллисекунды, поэтому Redis ставят рядом с основной базой данных для задач, где важна скорость:
- кеш — результаты медленных запросов к базе или внешнему API;
- счётчики и лимиты — просмотры, лайки, ограничение числа запросов в минуту;
- сессии и одноразовые коды — данные, которые должны исчезнуть через заданное время;
- очереди задач — брокер для Celery, RQ и собственных воркеров;
- рейтинги — таблицы лидеров на отсортированных множествах;
- обмен сообщениями — pub/sub между процессами и сервисами;
- распределённые блокировки — чтобы задачу выполнял только один процесс.
Redis умеет сохранять данные на диск (снимки RDB и журнал AOF), но обычно его используют как вспомогательное хранилище: если ключ пропал, приложение берёт данные из основной базы.
Redis или Valkey: какой сервер выбрать
В 2024 году Redis сменил лицензию, и у него появился форк Valkey. На код на Python это почти не влияет: клиент redis-py распространяется по лицензии MIT и работает с обоими серверами.
| Сервер | Лицензия | Что важно знать |
|---|---|---|
| Redis 7.2 и старше | BSD-3-Clause | последние версии с открытой лицензией BSD |
| Redis 7.4 | RSALv2 или SSPLv1 | лицензии с исходным кодом, но не открытые по критериям OSI |
| Redis 8 и новее | RSALv2, SSPLv1 или AGPLv3 на выбор | вернулась открытая лицензия AGPLv3; в сборку входят JSON, поиск и другие бывшие модули Redis Stack |
| Valkey | BSD-3-Clause | форк Redis 7.2 под управлением Linux Foundation, совместим по протоколу и командам |
Для изучения, локальной разработки и обычного использования в своём проекте подходит любой вариант. Лицензия важна компаниям, которые продают Redis как облачный сервис или меняют его исходный код. Часть облачных провайдеров предлагает Valkey вместо Redis. Для redis-py разницы нет, кроме команд, которые есть только в Redis 8, — например, типа JSON.
Как установить сервер Redis
redis-py — только клиент, ему нужен запущенный сервер. По умолчанию сервер слушает порт 6379.
Docker: любая ОС
Самый быстрый способ — контейнер. Команды одинаковы для Windows (Docker Desktop), macOS и Linux:
docker run -d --name redis -p 6379:6379 redis:8
docker exec -it redis redis-cli ping
PONG
Чтобы данные переживали удаление контейнера, добавьте том: -v redis-data:/data. Для Valkey замените образ на valkey/valkey:9, а redis-cli на valkey-cli. Остановить сервер — docker stop redis, запустить снова — docker start redis.
Windows: WSL или Docker
Официальной сборки Redis для Windows нет. Для разработки есть три пути:
-
Docker Desktop — команда из предыдущего раздела.
-
WSL (подсистема Windows для Linux) — Redis ставится внутри Ubuntu как на обычный Linux. В WSL 2 порт пробрасывается в Windows, поэтому Python на Windows подключается к
localhost:6379:wsl --install # один раз, затем перезагрузка# внутри Ubuntu в WSL sudo apt update sudo apt install redis-server sudo service redis-server start redis-cli ping # → PONG -
Memurai — совместимый с Redis сервер для Windows от партнёра Redis.
Важно: не используйте старый порт Redis от Microsoft с GitHub, который встречается в инструкциях: его последний релиз вышел в 2016 году (Redis 3.x), многие команды и параметры из этой статьи там не работают.
macOS и Linux
# macOS (Homebrew)
brew install redis
brew services start redis
# Ubuntu / Debian
sudo apt install redis-server
sudo systemctl enable --now redis-server
В репозиториях дистрибутивов часто лежит Redis 7.x — для примеров из статьи этого достаточно.
Установка redis-py
Пакет на PyPI называется redis (не redis-py). Ставьте его в виртуальное окружение проекта:
# Windows
py -m pip install redis
# macOS / Linux
python3 -m pip install redis
Проверка версии:
import redis
print(redis.__version__) # → 8.1.0
redis-py 8 требует Python 3.10 или новее. Дополнение redis[hiredis] ставит парсер ответов на C — он ускоряет разбор больших ответов, код при этом не меняется: pip install "redis[hiredis]". Подробнее об установке пакетов — в руководстве по pip.
Библиотека aioredis устарела: её код вошёл в redis-py как модуль redis.asyncio (с версии 4.2). Для нового кода ставьте только redis.
Подключение к Redis
Основной класс — redis.Redis. Значения по умолчанию: host="localhost", port=6379, db=0, без пароля.
import redis
r = redis.Redis(
host="localhost",
port=6379,
db=0, # номер базы, по умолчанию их 16: от 0 до 15
password=None, # пароль, если на сервере включён requirepass
decode_responses=True, # получать str вместо bytes
)
print(r.ping()) # → True
redis.Redis() не подключается к серверу при создании: соединение открывается при первой команде. Поэтому проверяйте доступность сервера вызовом r.ping() при старте приложения.
Строка подключения: redis:// и rediss://
Параметры удобно хранить в одной переменной окружения и передавать в from_url():
import os
import redis
# redis://[[пользователь]:пароль@]хост:порт/номер_базы
url = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
r = redis.Redis.from_url(url, decode_responses=True)
Схема rediss:// (с двумя s) включает TLS — её обычно выдают облачные провайдеры: rediss://default:пароль@redis.example.com:6380/0. Параметры клиента можно передать и в самой строке: redis://localhost:6379/0?socket_timeout=30.
decode_responses: байты или строки
По умолчанию redis-py возвращает bytes — так он не гадает, текст в ключе или двоичные данные:
raw = redis.Redis()
raw.set("greeting", "Привет, Redis")
print(raw.get("greeting")) # → b'\xd0\x9f\xd1\x80\xd0\xb8\xd0\xb2\xd0\xb5\xd1\x82, Redis'
r = redis.Redis(decode_responses=True)
print(r.get("greeting")) # → Привет, Redis
С decode_responses=True все ответы декодируются из UTF-8 в str. Если в Redis лежат картинки, сжатые данные или pickle, создайте для них отдельный клиент без этого параметра: при попытке декодировать двоичные данные будет UnicodeDecodeError.
Пул соединений и потоки
Каждый клиент Redis держит внутри пул соединений: открытые TCP-соединения переиспользуются, а один объект клиента безопасно вызывать из нескольких потоков. Создайте клиент один раз на модуль или приложение и импортируйте его, а не открывайте новый на каждую операцию.
Явный пул нужен, когда несколько клиентов должны делить общий лимит соединений. Параметры соединения в этом случае задают пулу:
pool = redis.ConnectionPool(host="localhost", port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)
Важно: если передать
decode_responses=Trueвredis.Redis(connection_pool=pool), а пулу — нет, параметр молча игнорируется иget()вернёт байты. Это частая причина вопроса «почему вместо строки приходитb'...'».
Таймауты и повторы в redis-py 8
В redis-py 8 изменились значения по умолчанию: socket_timeout и socket_connect_timeout — 5 секунд, до 100 соединений в пуле, 10 повторов при ошибке соединения с нарастающей паузой. Из-за повторов недоступный сервер обнаруживается не сразу: на Windows ошибка ConnectionError появляется только через 25–50 секунд. Для скриптов и проверок при старте отключите повторы:
import redis
from redis.backoff import NoBackoff
from redis.retry import Retry
r = redis.Redis(host="127.0.0.1", retry=Retry(NoBackoff(), 0))
try:
r.ping()
except redis.exceptions.ConnectionError as e:
print(e)
Error 10061 connecting to 127.0.0.1:6379. No connection could be made because the target machine actively refused it.
На Linux и macOS та же ситуация выглядит как Error 111 connecting to localhost:6379. Connection refused. Ошибка значит, что сервер не запущен или слушает другой порт.
Строки и счётчики
Строка — базовый тип Redis: значение до 512 МБ по ключу. Ключи принято называть с двоеточиями: user:42:name, cache:product:7.
r = redis.Redis(decode_responses=True)
r.set("user:1:name", "Анна") # → True
r.get("user:1:name") # → 'Анна'
r.get("user:404:name") # → None, ключа нет
r.mset({"user:1:name": "Анна", "user:2:name": "Олег"})
r.mget(["user:1:name", "user:2:name", "user:3:name"])
# → ['Анна', 'Олег', None]
r.exists("user:1:name", "user:404:name") # → 1, сколько ключей существует
r.delete("user:1:name") # → 1, сколько ключей удалено
Счётчики работают атомарно: два процесса, одновременно вызвавшие incr(), не потеряют ни одного увеличения.
r.incr("page:views") # → 1
r.incr("page:views") # → 2
r.incrby("page:views", 10) # → 12
r.decr("page:views") # → 11
r.get("page:views") # → '11' — строка, приведите через int()
Параметр nx=True записывает ключ, только если его ещё нет, — так делают простую защиту от повторного запуска:
r.set("job:report", "running", nx=True) # → True, ключ создан
r.set("job:report", "running", nx=True) # → None, ключ уже есть
Какие значения можно записать
Redis хранит только строки и байты. redis-py принимает str, bytes, int и float, а остальное отклоняет:
r.set("user:1", {"name": "Анна"})
redis.exceptions.DataError: Invalid input of type: 'dict'. Convert to a bytes, string, int or float first.
Та же ошибка будет для None и bool. Словари и списки сериализуйте в JSON — подробнее о модуле json в статье про JSON в Python:
import json
profile = {"name": "Анна", "tags": ["python", "redis"], "age": 28}
r.set("user:42:profile", json.dumps(profile, ensure_ascii=False), ex=3600)
data = r.get("user:42:profile")
profile = json.loads(data) if data else None
print(profile) # → {'name': 'Анна', 'tags': ['python', 'redis'], 'age': 28}
Время жизни ключей: TTL
TTL (time to live) — сколько секунд осталось жить ключу. Когда время выходит, Redis удаляет ключ сам. На этом строятся кеш, одноразовые коды и сессии.
from datetime import timedelta
r.set("code:79001234567", "4821", ex=300) # 5 минут
r.set("cache:rates", "...", ex=timedelta(hours=1)) # можно передать timedelta
r.set("token:abc", "...", px=1500) # в миллисекундах
r.ttl("code:79001234567") # → 300
r.ttl("user:2:name") # → -1, ключ без срока жизни
r.ttl("user:404:name") # → -2, ключа нет
r.expire("user:2:name", 60) # → True, назначить срок существующему ключу
r.persist("user:2:name") # → True, снять срок
Важно: обычный
set()безexсбрасывает TTL — ключ становится вечным. Чтобы обновить значение и сохранить оставшийся срок, передайтеkeepttl=True.
Метод setex() устарел: вместо r.setex(key, 300, value) пишите r.set(key, value, ex=300).
Типичный кеш медленной функции выглядит так:
import json
def get_product(product_id: int) -> dict:
key = f"cache:product:{product_id}"
cached = r.get(key)
if cached is not None:
return json.loads(cached)
product = load_product_from_db(product_id) # медленный запрос
r.set(key, json.dumps(product, ensure_ascii=False), ex=600)
return product
Хэши: объект с полями
Хэш — словарь внутри одного ключа. В нём удобно хранить профиль пользователя или настройки: каждое поле читается и меняется отдельно, без перезаписи всего объекта.
r.hset("user:42", mapping={"name": "Анна", "email": "anna@example.com", "visits": 0})
# → 3, сколько полей добавлено
r.hget("user:42", "name") # → 'Анна'
r.hincrby("user:42", "visits", 1) # → 1
r.hset("user:42", "city", "Казань") # одно поле
r.hdel("user:42", "email") # → 1
r.hgetall("user:42")
# → {'name': 'Анна', 'visits': '1', 'city': 'Казань'}
r.hmget("user:42", ["name", "city"]) # → ['Анна', 'Казань']
r.hgetall("user:404") # → {}, пустой словарь вместо None
Значения полей возвращаются строками, даже если записывали числа. Вложенные структуры хэш не поддерживает: mapping={"tags": ["a", "b"]} вызовет DataError — такое поле сериализуйте в JSON.
Метод hmset() устарел с redis-py 4.0 и выдаёт DeprecationWarning, используйте hset(name, mapping=...).
Списки: очередь задач
Список — упорядоченная последовательность строк. Добавлять можно слева (lpush) и справа (rpush), забирать — с любого конца. Пара rpush + lpop даёт очередь «первым пришёл — первым ушёл».
r.rpush("queue:emails", "a@example.com", "b@example.com") # → 2, длина списка
r.lpush("queue:emails", "vip@example.com") # → 3, в начало
r.lrange("queue:emails", 0, -1)
# → ['vip@example.com', 'a@example.com', 'b@example.com']
r.llen("queue:emails") # → 3
r.lpop("queue:emails") # → 'vip@example.com'
r.rpop("queue:emails") # → 'b@example.com'
ltrim() обрезает список — так хранят «последние N событий»:
r.lpush("recent:views", "item9")
r.ltrim("recent:views", 0, 99) # оставить 100 последних
Воркер очереди ждёт задачи блокирующим blpop(): он возвращает кортеж (ключ, значение) или None, если за timeout секунд ничего не пришло.
worker = redis.Redis(decode_responses=True, socket_timeout=None)
while True:
item = worker.blpop(["queue:emails"], timeout=30)
if item is None:
continue # очередь пуста, ждём дальше
_, email = item # → ('queue:emails', 'a@example.com')
send_email(email)
Важно: в redis-py 8
socket_timeoutпо умолчанию 5 секунд. Еслиblpop()ждёт дольше, клиент оборвёт ожидание сTimeoutError. Для воркера создайте отдельный клиент сsocket_timeout=Noneили держитеtimeoutменьшеsocket_timeout.
Для настоящих фоновых задач с повторами и расписанием берите готовую очередь — например, Celery с Redis в роли брокера, о нём статья про Celery в Django.
Множества и отсортированные множества
Множество хранит уникальные строки без порядка: кто лайкнул пост, какие теги у статьи, кто был онлайн сегодня.
r.sadd("post:1:likes", "anna", "oleg", "ivan") # → 3
r.sadd("post:1:likes", "anna") # → 0, уже есть
r.sismember("post:1:likes", "oleg") # → 1
r.scard("post:1:likes") # → 3, размер
r.smembers("post:1:likes") # → {'anna', 'ivan', 'oleg'}
r.sadd("post:2:likes", "oleg", "maria")
r.sinter("post:1:likes", "post:2:likes") # → {'oleg'}, пересечение
r.srem("post:1:likes", "ivan") # → 1
smembers() возвращает Python-множество set, порядок элементов не гарантирован. sismember() возвращает 1 или 0 — в условиях это работает как True и False.
Отсортированное множество (sorted set) добавляет каждому элементу числовой вес. Это готовая таблица лидеров:
r.zadd("leaderboard", {"anna": 120, "oleg": 95, "ivan": 150})
r.zincrby("leaderboard", 40, "oleg") # → 135.0
r.zrange("leaderboard", 0, 2, desc=True, withscores=True)
# → [('ivan', 150.0), ('oleg', 135.0), ('anna', 120.0)]
r.zrevrank("leaderboard", "anna") # → 2, место с нуля
r.zscore("leaderboard", "anna") # → 120.0
Какую структуру выбрать
| Задача | Структура | Основные методы |
|---|---|---|
| Кеш, флаг, одноразовый код | строка + TTL | set(ex=...), get, delete |
| Счётчик, лимит запросов | строка | incr, incrby, expire |
| Объект с отдельно изменяемыми полями | хэш | hset(mapping=...), hget, hgetall |
| Очередь, последние N событий | список | rpush, lpop, blpop, ltrim |
| Уникальные значения, теги, лайки | множество | sadd, sismember, sinter |
| Рейтинг, сортировка по весу | отсортированное множество | zadd, zincrby, zrange |
| Вложенный документ | JSON (Redis 8) | r.json().set, r.json().get |
Поиск ключей: scan_iter вместо keys
r.keys("cache:*") возвращает все подходящие ключи за один вызов, но на время поиска блокирует сервер. На базе с миллионами ключей это остановит все остальные запросы. В рабочем коде перебирайте ключи порциями через scan_iter():
for key in r.scan_iter(match="cache:*", count=500):
r.delete(key)
count — подсказка серверу, сколько ключей просматривать за шаг. Порядок ключей не гарантирован, а один ключ иногда встречается дважды.
Pipeline: несколько команд за один запрос
Каждая команда — отдельный сетевой запрос. Pipeline (конвейер) накапливает команды и отправляет их одной пачкой; execute() возвращает список ответов в том же порядке:
pipe = r.pipeline()
pipe.set("user:1:name", "Анна")
pipe.incr("stats:signups")
pipe.expire("stats:signups", 3600)
pipe.get("user:1:name")
print(pipe.execute()) # → [True, 1, True, 'Анна']
Команды можно записывать цепочкой: r.pipeline().set("a", 1).incr("a").get("a").execute() вернёт [True, 2, '2']. При массовых операциях — загрузке тысяч ключей или чтении сотни хэшей — pipeline сокращает время в разы, потому что убирает ожидание сети между командами.
with r.pipeline(transaction=False) as pipe:
for user_id in range(1, 101):
pipe.hgetall(f"user:{user_id}")
users = pipe.execute() # список из 100 словарей
Транзакции и WATCH
По умолчанию pipeline() оборачивает команды в транзакцию MULTI/EXEC: другие клиенты не вклинятся между ними. transaction=False отключает обёртку — так быстрее, когда атомарность не нужна.
Если решение зависит от текущего значения (списать деньги, только если хватает баланса), используйте watch(). Redis отменит транзакцию, если ключ изменился после watch(), и redis-py выбросит WatchError — операцию повторяют:
def withdraw(user: str, amount: int) -> bool:
key = f"balance:{user}"
with r.pipeline() as pipe:
while True:
try:
pipe.watch(key)
balance = int(pipe.get(key) or 0) # после watch команды выполняются сразу
if balance < amount:
pipe.unwatch()
return False
pipe.multi() # дальше — команды транзакции
pipe.decrby(key, amount)
pipe.execute()
return True
except redis.WatchError:
continue # ключ изменили, пробуем снова
r.set("balance:anna", 100)
print(withdraw("anna", 30), r.get("balance:anna")) # → True 70
print(withdraw("anna", 100)) # → False
Блокировка lock
Когда задачу должен выполнять только один процесс из нескольких (например, ночной отчёт на трёх серверах), подойдёт встроенная распределённая блокировка:
lock = r.lock("lock:nightly-report", timeout=300, blocking_timeout=5)
with lock:
build_report()
timeout — через сколько секунд блокировка снимется сама, если процесс упал. blocking_timeout — сколько ждать чужую блокировку; если не дождались, with выбросит redis.exceptions.LockError.
Pub/Sub: публикация и подписка
Pub/Sub рассылает сообщения всем подписчикам канала. Отправитель вызывает publish(), получатели подписываются через объект pubsub():
p = r.pubsub()
p.subscribe("notifications")
print(p.get_message(timeout=1.0))
# → {'type': 'subscribe', 'pattern': None, 'channel': 'notifications', 'data': 1}
r.publish("notifications", "Новый заказ #1024") # → 1, число получателей
print(p.get_message(timeout=1.0))
# → {'type': 'message', 'pattern': None, 'channel': 'notifications', 'data': 'Новый заказ #1024'}
Первым приходит служебное подтверждение подписки. С r.pubsub(ignore_subscribe_messages=True) такие сообщения пропускаются, но get_message() на их месте вернёт None — учитывайте это в цикле. Подписка по шаблону — p.psubscribe("orders:*"), тогда у сообщения type равен 'pmessage'.
Подписчик обычно работает в отдельном процессе или потоке и читает сообщения бесконечным циклом listen():
p = r.pubsub(ignore_subscribe_messages=True)
p.subscribe("chat")
for message in p.listen():
print("получено:", message["data"])
if message["data"] == "stop":
break
p.close()
Другой вариант — обработчик и фоновый поток:
def handler(message):
print(message["channel"], message["data"])
p = r.pubsub(ignore_subscribe_messages=True)
p.subscribe(alerts=handler)
thread = p.run_in_thread(sleep_time=0.1, daemon=True)
# ...
thread.stop()
Важно: Pub/Sub не хранит сообщения. Если подписчик в момент
publish()был отключён, сообщение потеряно —publish()вернёт0. Для надёжной доставки используйте списки или потоки Redis Streams (xadd,xreadgroup). Объектыpubsub()иpipeline()нельзя передавать между потоками: в отличие от клиента, их создают отдельно в каждом потоке.
Асинхронный клиент redis.asyncio
В асинхронных приложениях (FastAPI, aiohttp, боты на aiogram) используйте redis.asyncio — у него те же методы, только с await. Если async/await пока незнакомы, начните со статьи про асинхронность в Python.
import asyncio
import redis.asyncio as redis
async def main() -> None:
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
await r.set("greeting", "Привет", ex=60)
print(await r.get("greeting")) # → Привет
async with r.pipeline() as pipe:
pipe.incr("hits").hset("user:1", mapping={"name": "Анна"})
print(await pipe.execute()) # → [1, 1]
# несколько команд параллельно
print(await asyncio.gather(*(r.incr(f"counter:{i}") for i in range(3)))) # → [1, 1, 1]
await r.aclose()
asyncio.run(main())
Методы pipeline вызываются без await — он нужен только для execute(). Закрывайте клиент через await r.aclose() или async with redis.Redis(...) as r:. Метод close() в асинхронном клиенте устарел.
Подписка в асинхронном коде:
async def listen_news(r: redis.Redis) -> None:
pubsub = r.pubsub()
await pubsub.subscribe("news")
async for message in pubsub.listen():
if message["type"] == "message":
print("получено:", message["data"])
await pubsub.aclose()
Как правило, клиент создают один раз при старте приложения и закрывают при остановке — в FastAPI это делают в обработчике жизненного цикла lifespan. Основы фреймворка — в руководстве по FastAPI.
Частые ошибки
| Ошибка или симптом | Причина и решение |
|---|---|
ConnectionError: Error 10061 (Windows) или Error 111 (Linux) | сервер не запущен или слушает другой порт — проверьте docker ps или redis-cli ping |
| Ошибка появляется через полминуты | redis-py 8 повторяет подключение 10 раз; для быстрого отказа передайте retry=Retry(NoBackoff(), 0) |
Вместо строки b'...' | нет decode_responses=True или он передан клиенту, а не явному пулу |
DataError: Invalid input of type: 'dict' | Redis хранит только строки и числа — сериализуйте в JSON |
ResponseError: WRONGTYPE Operation against a key holding the wrong kind of value | команда для другого типа: get() по ключу-хэшу; тип ключа покажет r.type(key) |
TimeoutError в blpop() | ожидание дольше socket_timeout (5 с) — клиент с socket_timeout=None |
AuthenticationError | на сервере задан пароль — передайте password или укажите его в URL |
ModuleNotFoundError: No module named 'redis' | пакет установлен в другой интерпретатор — ставьте через python -m pip install redis в активном окружении |
Частые вопросы
Как сохранить словарь или объект в Redis?
Есть три варианта. JSON-строка через json.dumps() — универсально и читается из любого языка. Хэш через hset(mapping=...) — если поля нужно читать и менять по одному, но вложенность он не поддерживает. Тип JSON в Redis 8 — r.json().set("user:1", "$", profile) и r.json().get("user:1") — когда нужно менять вложенные поля на сервере. Pickle лучше не использовать: десериализация данных, которые мог подменить кто-то другой, позволяет выполнить произвольный код.
Как проверить код без запущенного Redis?
В тестах подойдёт библиотека fakeredis: fakeredis.FakeRedis(decode_responses=True) повторяет интерфейс redis.Redis и хранит данные в памяти процесса. Для асинхронного кода есть fakeredis.FakeAsyncRedis. Интеграционные тесты всё равно стоит гонять на настоящем сервере в Docker — эмулятор поддерживает не все команды и модули.
Чем redis.Redis отличается от redis.StrictRedis?
Ничем: StrictRedis — старое имя того же класса, оставленное для совместимости. В новом коде пишите redis.Redis.
Сколько баз данных в Redis и стоит ли их использовать?
По умолчанию 16 баз с номерами от 0 до 15, номер передаётся в параметре db или в конце URL: redis://localhost:6379/1. Отдельные базы удобны, чтобы развести кеш и очередь задач в одном локальном сервере. Кластерные и облачные установки часто разрешают только базу 0, поэтому в больших проектах данные разделяют префиксами ключей.
Как использовать Redis в Django?
В Django Redis подключают как бэкенд кеша и хранилище сессий через настройку CACHES — без прямых вызовов redis-py. Настройка, кеширование представлений и сессии разобраны в статье про Redis в Django.
Итог
Для Redis в Python достаточно пакета redis: создайте один клиент redis.Redis(..., decode_responses=True) или from_url() и вызывайте методы с именами команд Redis. Сервер для разработки проще всего запустить в Docker; на Windows — в Docker Desktop или WSL. Для кеша и кодов используйте set(..., ex=...), для объектов — хэши или JSON, для очередей — списки с blpop(), для групп команд — pipeline(), для асинхронных приложений — redis.asyncio. При переходе на redis-py 8 учтите новые таймауты и повторы по умолчанию. Другие популярные пакеты собраны в обзоре библиотек Python.
- #Python
- #Redis
- #redis-py
- #Кеширование
- #asyncio