Ich spiele jetzt schon etwas länger mit Python & Redis herum. Mittlerweile läuft die Redis DB lokal in einem Docker Container, die Daten werden regelmäßig gesichert. Ein gute Hilfe beim Spielen ist Redis Insight.
Die Datenbank sieht so aus.
Heute mal etwas Beispiel-Code. Die Imports, die wir benötigen
import redis
import dataclasses
import argparse
Funktion zur Initialisierung der Redis DB
###############################################
# Function for init Redis DB
###############################################
def init_redis(redis_host, redis_port, redis_db):
"""init redis db"""
redis_obj = redis.Redis(host=redis_host, port=redis_port, db=redis_db)
return redis_obj
Umschaltung Live und Test DB
Ich arbeite mit Live DB und Test DB. Hier definiere ich die verschiedenen Datenbanken, die ich benutze.
LIVE = [0, 1, 2] # redis db numbers
TEST = [10, 11, 12] # redis db numbers
Sollte selbsterklärend sein. Über eine Option beim Programmaufruf wird das gesteuert.
###############################################
# Argument parsing setup
###############################################
parser = argparse.ArgumentParser(description='Start the app with optional test mode.')
parser.add_argument('--test_mode', type=int, default=0, help='Set test mode (default: 0)')
args = parser.parse_args()
Programmaufruf
python3 main.py --test_mode
Der Python Code zur Umschaltung
#####################
# Loads DB depending on Test Mode
#####################
# Initialize connection
if args.test_mode == 0:
settings_redis_db = init_redis(SERVER_IP, 6379, LIVE[0])
else:
settings_redis_db = init_redis(SERVER_IP, 6379, TEST[0])
DB Klasse
@dataclasses.dataclass
class PortfolioSettings:
"""Class to handle Portfolio Settings.
Data is stored in Redis DB which runs in a docker container.
"""
project: str
version: str
theme: str
stock_list: str
def __init__(self, project="Portfolio", version="0.0.3", theme="dark", stock_list=None):
self.project = project
self.version = version
self.theme = theme
self.stock_list = stock_list if stock_list is not None else ["AAPL", "GOOG"]
def load_from_db(self, db_client):
"""Load settings from Redis DB."""
try:
self.project = str(db_client.json().get('settings', '$..project')[0]) or self.project
self.version = str(db_client.json().get('settings', '$..version')[0]) or self.version
self.theme = str(db_client.json().get('settings', '$..theme')[0]) or self.theme
stock_list = str(db_client.json().get('settings', '$..stock_list')[0])
self.stock_list = stock_list if stock_list is not None else self.stock_list
except (KeyError, Exception) as e: # Catching a generic exception here since we removed the specific RedisError
print("Error:", e)
def save_to_db(self, db_client):
"""Save settings to Redis DB."""
try:
data_to_save = {
'project': self.project,
'version': self.version,
'theme': self.theme,
'stock_list': self.stock_list
}
db_client.json().set('settings', '$', data_to_save)
except Exception as e: # Catching a generic exception
print("Error:", e)
def update(self, key, value):
"""Update a value."""
setattr(self, key, value)
settings_data = PortfolioSettings()
settings_data.load_from_db(settings_redis_db)
settings_data.save_to_db(settings_redis_db)
Init
Init legt die DB an, wenn keine DB vorhanden ist. Also, keine DB vorhanden, dann wird sie mit den hinterlegten Daten angelegt.
Load From DB
Wie der Name sagt, die Daten werden von der Redis DB geladen.
Save To DB
Wir speichern die Daten in der Datenbank.
Update
Ein Beispiel.
settings_data.update('theme', tab_settings['theme'])
Danach speichern nicht vergessen
settings_data.save_to_db(settings_redis_db)
Im Test-Modus sinnlos, da wir die DB beim Start neu initialisieren!
Und am Ende der Klasse stehen noch folgende beiden Zeilen.
settings_data = PortfolioSettings()
settings_data.load_from_db(settings_redis_db)
settings_data.save_to_db(settings_redis_db)
Das Objekt settings_data wird mit irgendwelchen Daten befüllt. Danach wird das Objekt settings_data mit den Daten aus der Redis DB befüllt. Wenn Sie nicht vorhanden ist, werden die Daten aus dem init angelegt.
Viel Spaß beim Testen und wie immer, wer Fehler findet den bitte ich um entsprechende Korrektur.
