Django-Admin.ru

Redis в Python: библиотека redis-py

Как работать с Redis в Python через redis-py: установка сервера и библиотеки, подключение, строки, хэши, списки, TTL, pipeline, pub/sub и redis.asyncio.

Редакция Django-Admin.ru
· 19 мин чтения

Содержание · 18 разделов
  1. Что такое Redis и для чего он нужен
  2. Redis или Valkey: какой сервер выбрать
  3. Как установить сервер Redis
  4. Docker: любая ОС
  5. Windows: WSL или Docker
  6. macOS и Linux
  7. Установка redis-py
  8. Подключение к Redis
  9. Строка подключения: redis:// и rediss://
  10. decode_responses: байты или строки
  11. Пул соединений и потоки
  12. Таймауты и повторы в redis-py 8
  13. Строки и счётчики
  14. Какие значения можно записать
  15. Время жизни ключей: TTL
  16. Хэши: объект с полями
  17. Списки: очередь задач
  18. Множества и отсортированные множества
  19. Какую структуру выбрать
  20. Поиск ключей: scan_iter вместо keys
  21. Pipeline: несколько команд за один запрос
  22. Транзакции и WATCH
  23. Блокировка lock
  24. Pub/Sub: публикация и подписка
  25. Асинхронный клиент redis.asyncio
  26. Частые ошибки
  27. Частые вопросы
  28. Как сохранить словарь или объект в Redis?
  29. Как проверить код без запущенного Redis?
  30. Чем redis.Redis отличается от redis.StrictRedis?
  31. Сколько баз данных в Redis и стоит ли их использовать?
  32. Как использовать Redis в Django?
  33. Итог

Для работы с 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.4RSALv2 или SSPLv1лицензии с исходным кодом, но не открытые по критериям OSI
Redis 8 и новееRSALv2, SSPLv1 или AGPLv3 на выборвернулась открытая лицензия AGPLv3; в сборку входят JSON, поиск и другие бывшие модули Redis Stack
ValkeyBSD-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

Какую структуру выбрать

ЗадачаСтруктураОсновные методы
Кеш, флаг, одноразовый кодстрока + TTLset(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
↑ ↓ выборEnter открытьEsc закрыть