redis-py ile Python'da Redis Kullanımı: Bağlantı, Pipeline ve Async

redis-py kurulumu ve hiredis desteği

pip install redis

# daha hızlı yanıt ayrıştırma için hiredis desteğiyle
pip install redis[hiredis]

İkinci komut, C ile yazılmış hiredis yanıt ayrıştırıcısını da kurar. Belgelere göre ortamda hiredis 1.0 veya üstü varsa redis-py yanıtları ayrıştırmak için onu kendiliğinden kullanmaya çalışır ve çoğu durumda kodunuzda hiçbir değişiklik gerekmez. Python 3.12 ve sonrasında distutils paketleme düzeni dilden çıkarıldığı için kurulumda sorun yaşarsanız belgeler güncel bir redis-py sürümüne geçmenizi önerir.

redis-py ile Redis'e bağlanma ve ilk komutlar

En basit bağlantı yerel sunucuya 6379 portundan yapılır. Belgelerdeki örnekte decode_responses=True kullanıldığı için get() doğrudan bar metnini döndürür:

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

r.set('foo', 'bar')
# True
r.get('foo')
# bar

Bir sözlüğü hash olarak saklamak için hset() metodunun mapping argümanını kullanırsınız. Geri okurken hgetall() bir Python sözlüğü döndürür. Çıktıda age alanının 29 sayısı yerine '29' metni olarak döndüğüne dikkat edin: Redis hash alanlarını metin olarak saklar, tip dönüşümü sizin sorumluluğunuzdadır.

r.hset('user-session:123', mapping={
    'name': 'John',
    "surname": 'Smith',
    "company": 'Redis',
    "age": 29
})

r.hgetall('user-session:123')
# {'surname': 'Smith', 'name': 'John', 'company': 'Redis', 'age': '29'}

Hash komutlarının kendisini Redis hash komutları yazısında, metin komutlarını SET, GET ve INCR yazısında ayrıntılı anlatıyoruz. redis-py tarafında genel kural basittir: Redis komutunun küçük harfli adı metot adıdır ve seçenekler anahtar sözcük argümanı olarak verilir. Örneğin belgelerdeki bir örnekte r.set(key, value, ex=3600) ile anahtar bir saatlik süreyle yazılır.

from_url() ile URL'den bağlanma

Sunucu, port, kimlik bilgileri, veritabanı ve TLS ayarını tek bir URL ile vermek için redis.from_url(), Redis.from_url() veya ConnectionPool.from_url() kullanılır; redis.asyncio modülünde de aynı fonksiyonlar bulunur.

import redis

r = redis.from_url("rediss://default:<password>@<host>:6379/0?protocol=3")

Bağlantı havuzu kullanımı

Üretimde bağlantıları tek tek açıp kapatmak yerine bir bağlantı havuzu kullanmanız önerilir. Havuzdan bir bağlantı aldığınızda havuz açık bağlantılarından birini verir; kapattığınızda bağlantı gerçekte kapanmaz, yeniden kullanılmak üzere havuza döner. Belgelerdeki örnek bir havuzu üç istemci arasında paylaşır:

import redis

pool = redis.ConnectionPool().from_url("redis://localhost")
r1 = redis.Redis().from_pool(pool)
r2 = redis.Redis().from_pool(pool)
r3 = redis.Redis().from_pool(pool)

r1.set("wind:1", "Hurricane")
r2.set("wind:2", "Tornado")
r3.set("wind:3", "Mistral")

r1.close()
r2.close()
r3.close()

pool.close()

Pratikte çoğu uygulamanın bu kadar ayrıntıya ihtiyacı yoktur. Asenkron rehberindeki açıklamaya göre bir Redis istemci nesnesi kendi bağlantı havuzunu zaten içeride oluşturup yönetir. Bu yüzden uzun süre çalışan bir uygulamada önerilen yol, açılışta tek bir istemci oluşturmak, onu istekler ve görevler arasında paylaşmak ve kapanışta kapatmaktır. Her istek için yeni bir Redis() oluşturmak havuzun amacını boşa çıkarır.

Cluster, Sentinel ve TLS bağlantıları

Redis Cluster'a bağlanmak için redis.cluster modülündeki RedisCluster sınıfı kullanılır. Belgelerdeki örnekte decode_responses verilmediği için get() sonucunun b'bar' biçiminde bytes döndüğünü görebilirsiniz:

from redis.cluster import RedisCluster

rc = RedisCluster(host='localhost', port=16379)

rc.set('foo', 'bar')
# True

rc.get('foo')
# b'bar'

Sentinel ile çalışırken Sentinel nesnesine Sentinel adreslerini verip master_for('mymaster') ile birincil sunucuya bağlanırsınız; salt okunur komutları kopyaya göndermek için slave_for() vardır. Sentinel'lerin ve veri düğümlerinin kimlik bilgileri ayrıdır: Sentinel şifresi sentinel_kwargs içinde, birincil sunucunun kullanıcı adı ve şifresi master_for() çağrısında verilir. Belgeler önemli bir sınırlamaya da dikkat çeker: redis-py Sentinel'lerin yük devretme duyurularını dinlemez, bu yüzden eski birincil yazmaları reddetmeye başlayana kadar onu kullanmaya devam eder ve o aralıkta kabul edilen yazmalar kaybolur. TLS için ssl=True ile birlikte ssl_certfile, ssl_keyfile ve ssl_ca_certs parametreleri kullanılır. Cluster'da çok anahtarlı komutların neden hata verebileceğini hash slot ve CROSSSLOT yazısında anlatıyoruz.

Pipeline ve transaction kullanımı

Pipeline, birden fazla komutu tek bir ağ iletişiminde gönderip tüm yanıtları tek seferde almanızı sağlar. redis-py'de önce r.pipeline() ile bir pipeline nesnesi oluşturur, komutları normal metot adlarıyla ona eklersiniz. Komutlar tamponda bekler ve yalnızca execute() çağrıldığında çalışır; execute() tüm sonuçları sırasıyla içeren bir liste döndürür. Pipeline metotları pipeline nesnesinin kendisini döndürdüğü için komutları zincirleyebilirsiniz de.

Belgelerdeki oturum deposu örneği bunun neden önemli olduğunu gösterir: oturum hash'i ile süresi aynı pipeline'da gönderildiği için anahtar hiçbir zaman süresiz kalmaz.

# Pipeline sends HSET and EXPIRE together so the key never exists without a TTL.
pipeline = r.pipeline()
pipeline.hset(key, mapping=payload)
pipeline.expire(key, session_ttl)
pipeline.execute()

redis-py'de pipeline varsayılan olarak bir transaction gibi çalışır, yani komutlar MULTI/EXEC içinde kesintisiz bir sıra olarak yürütülür. Bu davranışı kapatmak, yani yalnızca ağ gidiş dönüşünden tasarruf etmek istiyorsanız pipeline'ı transaction=False ile oluşturursunuz:

pipe = r.pipeline(transaction=False)

Pipelining'in sunucu tarafındaki mantığını ve betik kullanımıyla farkını Redis pipelining yazısında, transaction'ların hata türlerini ve geri alma olmamasını MULTI, EXEC ve WATCH yazısında anlatıyoruz.

WATCH ile iyimser kilit ve WatchError

Bir değeri okuyup ona göre güncellediğiniz durumlarda, arada başka bir istemci aynı anahtarı değiştirirse güncellemeniz eski veriye dayanır. Redis bunu iyimser kilit (optimistic locking) ile çözer: anahtarı izlemeye alırsınız, değişmişse transaction iptal edilir ve redis-py WatchError fırlatır; siz de baştan denersiniz. Belgelerdeki önbellek örneği yalnızca önbellekte zaten bulunan bir kaydın tek alanını günceller:

import redis

def update_field(r, cache_key, field, value, ttl):
    with r.pipeline() as pipe:
        while True:
            try:
                pipe.watch(cache_key)
                if not pipe.exists(cache_key):
                    pipe.unwatch()
                    return False
                pipe.multi()
                pipe.hset(cache_key, field, value)
                pipe.expire(cache_key, ttl)
                pipe.execute()
                return True
            except redis.WatchError:
                continue

Akışa dikkat edin: watch() çağrısından sonra pipeline anlık moddadır, yani exists() sonucu hemen döner. multi() çağrıldıktan sonra komutlar yeniden tamponlanır ve execute() ile birlikte gönderilir. Belgeler bu kalıp yaygın olduğu için kütüphanede bir transaction() kolaylık metodu da bulunduğunu söyler: ona transaction kodunu içeren bir fonksiyonu ve izlenecek anahtarları verirsiniz, izleme, yürütme ve gerekirse yeniden deneme işini metot üstlenir. Ancak transaction() sizin yerinize multi() çağrısını ekleyemez; onu fonksiyonunuzda doğru yere kendiniz koymanız gerekir.

scan_iter() ile anahtarları güvenli tarama

SCAN, HSCAN, SSCAN ve ZSCAN komutları büyük sonuçları küçük sayfalar hâlinde döndürür. Komutları doğrudan kullanırken imleç değerini sizin takip etmeniz gerekir. redis-py bunun yerine sayfalamayı arka planda yöneten yineleyiciler sunar: scan_iter(), hscan_iter(), sscan_iter() ve zscan_iter(). Belgelerdeki örnekte sonuçların eklenme sırasıyla gelmediğine dikkat edin; SCAN komutunda olduğu gibi sıralama garanti edilmez:

import redis

r = redis.Redis(decode_responses=True)

r.set("key:1", "a")
r.set("key:2", "b")
r.set("key:3", "c")
r.set("key:4", "d")
r.set("key:5", "e")

for key in r.scan_iter():
    print(f"Key: {key}, value: {r.get(key)}")
# >>> Key: key:1, value: a
# >>> Key: key:4, value: d
# >>> Key: key:3, value: c
# >>> Key: key:2, value: b
# >>> Key: key:5, value: e

scan_iter() çağrısına match, count ve _type parametrelerini vererek dönen anahtarları daraltabilirsiniz. Diğer yineleyicilere ise taranacak nesnenin anahtarını verirsiniz. zscan_iter() her öğe için üye ve puandan oluşan iki elemanlı bir tuple döndürür; hscan_iter() varsayılan olarak alan ve değer ikilisi döndürür, yalnızca alan adlarını istiyorsanız no_values=True verirsiniz:

r.zadd("battles", mapping={
    "hastings": 1066,
    "agincourt": 1415,
    "trafalgar": 1805,
    "somme": 1916,
})

for item in r.zscan_iter("battles"):
    print(f"Key: {item[0]}, value: {int(item[1])}")
# >>> Key: hastings, value: 1066
# >>> Key: agincourt, value: 1415
# >>> Key: trafalgar, value: 1805
# >>> Key: somme, value: 1916

Üretim sunucusunda KEYS yerine neden SCAN tabanlı tarama tercih edilmesi gerektiğini ve SCAN'in hangi garantileri verdiğini SCAN ve KEYS farkı yazısında anlatıyoruz.

redis.asyncio ile asenkron kullanım

redis-py, redis.asyncio ad alanı altında asyncio uyumlu bir API sunar. Bu API senkron istemciyi birebir yansıtır; fark, komutları çağırmak yerine await etmenizdir. Belgeler asenkron istemciyi G/Ç ağırlıklı işler, FastAPI, Starlette, aiohttp veya Sanic gibi asenkron çatılarla entegrasyon ve tek süreçten çok sayıda eşzamanlı Redis işlemi çalıştırma durumları için önerir. Basit betikler, CPU ağırlıklı işler ya da olay döngüsü olmayan kod tabanları için senkron istemci genellikle daha iyi bir seçimdir.

Asenkron istemcinin oluşturulması senkrondur ve bağlantı açmaz; havuz ilk komutta bağlantıyı tembel biçimde kurar. İşiniz bittiğinde aclose() çağırmanız gerekir; eski close() metodu kullanımdan kaldırılmıştır. Belgelerin önerdiği yol, istemciyi async with ile bağlam yöneticisi olarak kullanmaktır; böylece bir istisna oluşsa bile kapatma çalışır. Hata yönetimi sayfasındaki örnek, bağlantı hatasında veritabanına dönen asenkron bir fonksiyonu gösterir:

import redis.asyncio as redis

async def get_with_fallback(key):
    r = await redis.from_url("redis://localhost")
    try:
        return await r.get(key)
    except redis.ConnectionError:
        logger.warning("Cache unavailable")
        return await database.get(key)
    finally:
        await r.aclose()

Senkron bir örneği asenkrona çevirmek için belgeler şu kuralları verir:

• import redis yerine import redis.asyncio as redis yazın ve örneği asyncio.run(...) ile çağrılan bir async def içine alın.
• Her komut çağrısının önüne await ekleyin. İstisna: pipeline'a tamponlanan pipe.set(...) gibi komutlar await edilmez; yalnızca pipe.watch(...), multi() öncesindeki okumalar ve pipe.execute() await edilir. multi() yalnızca iç durumu değiştirdiği için senkron kalır.
• with r.pipeline(...) as pipe: yerine async with r.pipeline(...) as pipe: kullanın.
• r.close() yerine await r.aclose() yazın ya da istemciyi async with ile açın.

Asenkron kodda en sık yapılan hata await yazmayı unutmaktır: bu durumda sonuç yerine bir coroutine nesnesi alırsınız. Ayrıca çalışmakta olan her komut havuzdan bir bağlantı tüketir; bu yüzden max_connections değerini beklediğiniz en yüksek eşzamanlılığa göre ayarlayın. Havuz tükendiğinde hata almak yerine beklemek istiyorsanız istemciyi BlockingConnectionPool ile oluşturabilirsiniz. Redis istemcisi görevler arasında paylaşılabilir, fakat ondan türetilen pipeline ve PubSub nesneleri paylaşılamaz; her görev kendi nesnesini almalıdır. Bir komutu asyncio.wait_for() ile iptal ederseniz redis-py, sonraki okumalarda istek ve yanıtların karışmaması için o bağlantıyı koparır ve sonraki komut havuzdan yeni bir bağlantı alır.

redis-py istisna hiyerarşisi

redis-py hataları istisnalarla bildirir. Temel sınıf redis.RedisError'dır; ConnectionError bunun alt sınıfıdır ve onun altında da TimeoutError ile BusyLoadingError bulunur. Diğer alt sınıflar arasında ResponseError, InvalidResponse, DataError ve PubSubError vardır. Kimlik doğrulama hataları için ConnectionError'ın alt sınıfı olan AuthenticationError, izlenen anahtar değiştiğinde ise WatchError kullanılır.

Belgeler ResponseError için üç tipik örnek verir: string tutan bir anahtara LPUSH göndermek, zaten var olan bir adla arama indeksi oluşturmak ve stream girdisi için geçersiz bir ID kullanmak. Bunlar koddaki hatalardır; yeniden denemek işe yaramaz. Doğru yaklaşım hata türlerini ayırmaktır:

try:
    result = r.get(key)
except redis.TimeoutError:
    retry()  # Retry on timeout
except redis.ResponseError:
    raise   # Fail on syntax error

Sunucu hata önekleri bazı kütüphanelerde kendi sınıflarına dönüşür. Örneğin belgelerin hata mesajları sayfasına göre redis-py, sunucudan gelen READONLY hatasını ReadOnlyError olarak bildirir ve mesajdaki öneki atar. Bu hatanın sunucu tarafındaki anlamını OOM, MISCONF ve READONLY hataları yazısında bulabilirsiniz.

Üretim ayarları: yeniden deneme, sağlık kontrolü, zaman aşımı

Yeniden deneme. redis-py 6.0.0 sürümünden itibaren başarısız bir komutu varsayılan olarak üç kez dener. Denemeler arasındaki süre üstel geri çekilme (exponential backoff) ile hesaplanır ve bağlantıların aynı anda yeniden denemesini önlemek için rastgele bir sapma (jitter) eklenir. Bu davranışı değiştirmek için bağlantıda retry parametresine bir Retry nesnesi verirsiniz:

from redis import Redis
from redis.backoff import ExponentialBackoff
from redis.retry import Retry
from redis.exceptions import BusyLoadingError

# Run 8 retries with exponential backoff strategy.
retry = Retry(ExponentialBackoff(), 8)

r = Redis(
    retry=retry,
    retry_on_error=[BusyLoadingError],
    health_check_interval=3,
    socket_connect_timeout=15,
    socket_timeout=5,
)

Yeniden deneme, Retry sınıfının supported_errors özniteliğindeki istisnalardan biri oluştuğunda tetiklenir; varsayılan liste yalnızca ConnectionError ve TimeoutError içerir. Listeyi Retry(ExponentialBackoff(), 3, supported_errors=(TimeoutError,)) biçiminde kendiniz belirleyebilir ya da örnekteki gibi retry_on_error ile varsayılan listeye ekleme yapabilirsiniz. Cluster bağlantılarında retry nesnesi verilebilir ama istisna listesi değiştirilemez; her zaman TimeoutError, ConnectionError ve ClusterDownError'dır.

Sağlık kontrolü. health_check_interval saniye cinsinden bir tam sayı alır. Bağlantı bu süreden uzun boşta kalırsa redis-py bir sonraki komuttan önce kendiliğinden PING gönderip yanıtı kontrol eder. Sağlık kontrolleri de diğer komutlar gibi yeniden deneme stratejisine tabidir.

Zaman aşımı. Varsayılan olarak hem bağlantı kurma hem komutlar için zaman aşımı 10 saniyedir. socket_connect_timeout bağlantı, socket_timeout komut süresini belirler. Belgeler dengeye dikkat çeker: çok kısa süreler, biraz daha beklense başarılı olacak komutların gereksiz yere yeniden denenmesine yol açar; çok uzun süreler ise hiç gelmeyecek bir yanıt için uygulamanın askıda kalmasına neden olur.

İstemci tarafı önbellek ve izleme. Sık okunan anahtarları istemcide önbelleğe almak için bağlantıda protocol=3 ve cache_config=CacheConfig() verilir; bu özellik redis-py 5.1.0 veya üstünü ve en geniş uyumluluk için Redis 7.4 veya üstünü gerektirir. redis-py şu an yayın (broadcasting) ve opt-in/opt-out kiplerini uygulamaz. Konunun sunucu tarafı client-side caching yazısında. Performans ve hata metrikleri için redis-py OpenTelemetry destekler; gerekli paketler pip install redis[otel] ile kurulur.

Sık yapılan hatalar

• bytes beklemeden metin karşılaştırmak. decode_responses vermeden r.get('foo') == 'bar' yazarsanız sonuç yanlış olur, çünkü dönen değer b'bar''dır.
• Her istekte yeni istemci oluşturmak. Tek istemciyi uygulama genelinde paylaşın; havuz zaten içeride.
• Tek ConnectionPool'u birden fazla istemciye vermek. Birinin kapanması hepsini etkiler.
• Asenkron kodda await unutmak. Sonuç yerine coroutine nesnesi alırsınız.
• Pipeline ve PubSub nesnelerini görevler ya da iş parçacıkları arasında paylaşmak. Her iş birimi kendi nesnesini kullanmalıdır.
• Tüm istisnaları except Exception ile yakalamak. Bu, koddaki hataları gizler; belirli sınıfları yakalayın.

Sık Sorulan Sorular

redis-py neden b'...' biçiminde değer döndürüyor?

redis-py varsayılan olarak bütün yanıtları bytes olarak döndürür. Metin almak için istemciyi decode_responses=True ile oluşturun. Bu seçeneği URL'de değil, kodda verin.

Pipeline ile transaction aynı şey mi?

redis-py'de pipeline() varsayılan olarak transaction gibi çalışır ve komutları kesintisiz yürütür. Yalnızca ağ tasarrufu istiyorsanız pipeline(transaction=False) kullanın.

Asenkron istemciyi her istekte açıp kapatmalı mıyım?

Hayır. Belgeler uygulama açılışında tek bir istemci oluşturup paylaşmayı ve kapanışta aclose() ile kapatmayı önerir. FastAPI gibi çatılarda bunu açılış ve kapanış kancalarında yapabilirsiniz.

Bağlantı koptuğunda kendim yeniden deneme yazmalı mıyım?

Çoğu durumda gerekmez. redis-py 6.0.0 ve sonrasında başarısız komutları üç kez dener; sayıyı ve stratejiyi Retry nesnesiyle değiştirebilirsiniz.

Marka adları sahiplerine aittir. Bu yazı bağımsız bir teknik rehberdir; adı geçen kuruluşlarla bayilik veya ortaklık ilişkisi bulunduğu anlamına gelmez.

İlgili içerikler: Redis öğrenme rehberi · Redis bağlantı URL ve connection pool · Redis pipelining · MULTI, EXEC ve WATCH · SCAN ve KEYS farkı · Python öğrenme rehberi