Valkey: Naudojimas su Python

Kaip sujungti Valkey su Python

👋 Sveiki atvykę į Stackhero dokumentaciją!

Stackhero siūlo paruoštą naudoti Valkey cloud sprendimą, kuris suteikia daugybę privalumų, įskaitant:

Įtrauktą Valkey Admin web UI.

Neribotas žinučių dydis ir perdavimas.

Paprasti atnaujinimai vienu paspaudimu.

Optimali veikla ir stipri saugumo apsauga, užtikrinama naudojant privačią, dedikuotą infrastruktūrą.

Taupykite laiką ir supaprastinkite savo darbą: išbandyti Stackhero Valkey cloud hosting sprendimą užtrunka tik 5 minutes!

Python bibliotekos pasirinkimas Valkey integracijai

Norėdami sklandžiai integruoti savo programą su Valkey, apsvarstykite galimybę naudoti redis biblioteką Python. Ši biblioteka puikiai tinka jungtis prie Valkey, atvirojo kodo Redis šakos. Ją galite lengvai įdiegti vykdydami:

pip install redis
pip freeze > requirements.txt

Python prijungimas prie Valkey

Pažvelkime, kaip galite prijungti savo Python programą prie Valkey. Čia yra paprastas pavyzdys naudojant numatytuosius nustatymus:

import redis

r = redis.from_url(
  'rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>',
  health_check_interval=10,
  socket_connect_timeout=5,
  retry_on_timeout=True,
  socket_keepalive=True
)

Siekiant didesnio saugumo, rekomenduojama tvarkyti kredencialus naudojant aplinkos kintamuosius. Štai kaip tai padaryti:

import os
import redis

r = redis.from_url(
  os.environ.get("STACKHERO_VALKEY_URL_TLS"),
  health_check_interval=10,
  socket_connect_timeout=5,
  retry_on_timeout=True,
  socket_keepalive=True
)

Įsitikinkite, kad jūsų aplinkos kintamuosiuose yra tokia įrašas: STACKHERO_VALKEY_URL_TLS=rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>.

Pub/Sub naudojimas su Valkey ir Python

Naudotis Publish/Subscribe (Pub/Sub) funkcionalumu Python su Valkey yra paprasta. Štai pavyzdys:

import redis

# Prisijungti prie Valkey
r = redis.from_url(
  'rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>',
  health_check_interval=10,
  socket_connect_timeout=5,
  socket_keepalive=True,
  retry_on_timeout=True
)

# Sukurti PubSub instanciją
p = r.pubsub()

# Prenumeruoti kanalą "test"
p.subscribe('test')

# Paskelbti žinutę kanale "test"
r.publish('test', 'Tai yra testinė žinutė')

# Gauti pirmą prieinamą žinutę iš kanalo "test"
p.get_message()

# Atsisakyti prenumeratos kanale "test"
p.unsubscribe('test')

Pažangūs Valkey Pub/Sub pavyzdžiai su Python

Išplėskite savo Pub/Sub galimybes su šiais pažangiais pavyzdžiais:

# Sukurti PubSub instanciją ir ignoruoti prenumeratos žinutes
p = r.pubsub(ignore_subscribe_messages=True)

# Prenumeruoti kelis kanalus
p.subscribe('test-1', 'test-2', ...)

# Atsisakyti prenumeratos keliuose kanaluose
p.unsubscribe('test-1', 'test-2', ...)

# Taip pat galite naudoti "unsubscribe" be argumentų, kad atsijungtumėte nuo visų prenumeruotų kanalų
p.unsubscribe()

# Prenumeruoti kanalus naudojant šabloną
p.psubscribe('my-*')

# Atsisakyti prenumeratos kanaluose naudojant šabloną
p.punsubscribe('my-*')

Kaip išvengti "Connection Closed by Server" klaidų

Klaida redis.exceptions.ConnectionError: Connection closed by server gali atsirasti, jei jūsų programa nesąveikauja su Valkey tam tikrą laiką, dėl ko įvyksta automatinis atsijungimas. Norėdami to išvengti, galite nustatyti health_check_interval parametrą, kaip parodyta žemiau:

r = redis.from_url(
  'rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>',
  health_check_interval=10,
  socket_connect_timeout=5,
  retry_on_timeout=True,
  socket_keepalive=True
)

Naudojant Valkey Pub/Sub funkciją, redis-py biblioteka tikisi, kad funkcijos kaip get_message() ar listen() bus kviečiamos dažniau nei health_check_interval. Mūsų pavyzdyje šis intervalas nustatytas 10 sekundžių, todėl įsitikinkite, kad kviečiate get_message() ar listen() bent kartą kas 10 sekundžių (žr. redis-py oficialią dokumentaciją).

Jei to nepadarysite, galite susidurti su ta pačia prisijungimo klaida. Norėdami to išvengti, apsvarstykite galimybę reguliariai kviesti check_health().

Štai kaip galite tai įgyvendinti:

import redis
import threading

# Prisijungti prie Valkey
r = redis.from_url(
  'rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>',
  health_check_interval=10,
  socket_connect_timeout=5,
  retry_on_timeout=True,
  socket_keepalive=True
)

# Sukurti PubSub instanciją
p = r.pubsub()

# Prenumeruoti kanalą "test"
p.subscribe('test')

# Sukurti funkciją, kuri kvies `check_health` kas 5 sekundes
def valkey_auto_check(p):
  t = threading.Timer(5, valkey_auto_check, [p])
  t.start()
  p.check_health()

# Kviesti valkey_auto_check funkciją
valkey_auto_check(p)