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
###############################################
# 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
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])
@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 legt die DB an, wenn keine DB vorhanden ist. Also, keine DB vorhanden, dann wird sie mit den hinterlegten Daten angelegt.
Wie der Name sagt, die Daten werden von der Redis DB geladen.
Wir speichern die Daten in der Datenbank.
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.
Da kann man sich immer wieder mit beschäftigen und man lernt jeden Tag was dazu. Also, der Vorteil von Dataclasses ist, dass man viele Dinge nicht selber schreiben muss. So was wie _init() und repr()
This module provides a decorator and functions for automatically adding generated special methods such as init() and repr() to user-defined classes. It was originally described in PEP 557.
Kann man hier nachlesen. Ok, warum habe ich dann da oben eine init() mit drin? Kann das weg? Versuch macht klug.
@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 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)
# Initialisierung, falls Redis DB nicht vorhanden ist!
settings_data = PortfolioSettings(project="Portfolio", version="0.0.3", theme="dark", stock_list="['AAPL', 'GOOG']")
# Versuch die DB zu laden, wenn da werden Daten aus der DB geladen!
try:
settings_data.load_from_db(settings_redis_db)
except Exception as e: # Catching a generic exception
print("Error2:", e)
# Speichern, wenn keine DB vorhanden. Wenn vorhanden, kommt das hier 'Entry with key 0 already exists'.
settings_data.save_to_db(settings_redis_db)
Die letzten drei Befehle haben mir ein paar Kopfschmerzen verursacht. Jetzt habe ich es verstanden und mal entsprechend kommentiert.
Und schwupps, ist der init() Teil weg.
Ich bin mit der Lernkurve noch nicht so richtig zufrieden. Eine Frage die sich mir stellte, geht das einfacher? Der Ursprung meiner Datenbank Struktur liegt in einem anderen Projekt, wo ich versucht habe Daten permanent in einem File zu speichern. Dazu hatte ich damals JSON genommen. Deswegen auch diese Zeilen
self.project = str(db_client.json().get('settings', '$..project')[0]) or self.project
Gut, ich hatte dann mal ChatGPT gefragt, wie macht man das so 'normalerweise'? es kam eine Klasse heraus, die ich dann intensiv ausprobiert habe, ein wenig umgebaut usw. So lange, bis ich der Meinung war, ok ich habe es verstanden. Jetzt nutzte der Code auch mehr Redis Funktionen, wie
self.client.hset('settings', name, json.dumps(data))
Es waren jetzt folgende Funktionen drin
Dokumentation -> https://redis.io/commands/hset/
Beim Durchlesen des Codes hatte ich jetzt mehr das Gefühl, so muss das sein 
In RedisInsight sieht das dann jetzt so aus.
class PortfolioSettings:
def __init__(self, host='172.17.0.2', port=6379, db=0):
if args.test_mode == 1:
self.client = redis.StrictRedis(host=SERVER_IP, port=port, db=TEST[0])
else:
self.client = redis.StrictRedis(host=SERVER_IP, port=port, db=LIVE[0])
def set_settings(self, name, data):
"""Init settings if db don't exist"""
if not self.client.hexists('settings', name):
self.client.hset('settings', name, json.dumps(data))
return True
return False
def edit_setting(self, name, data):
"""Edit an entry in settings"""
if self.client.hexists('settings', name):
self.client.hset('settings', name, json.dumps(data))
return True
return False
def delete_setting(self, name):
"""Delete an entry in settings"""
return self.client.hdel('settings', name)
def get_setting(self, name):
"""Get an entry in settings"""
setting = self.client.hget('settings', name)
return json.loads(setting) if setting else None
def get_all_settings(self):
"""Get all entries in settings"""
settings = self.client.hgetall('settings')
return {k.decode(): json.loads(v) for k, v in settings.items()}
Und hier die Initialisierung
settings_data = PortfolioSettings()
#####################
# Will only be executed if DB is not available!
#####################
if not settings_data.get_all_settings():
# Settings initialisieren
print("INIT")
settings_data.set_settings("project", "Portfolio")
settings_data.set_settings("version", "0.0.3")
settings_data.set_settings("theme", "dark")
settings_data.set_settings("url_list", ["https://www.onvista.de/aktien/Deutsche-Telekom-Aktie-DE0005557508"])
settings_data.set_settings("exchange_list", ['DKB','Smartbroker','BUX'])
Teile der Klasse sind [KI-generiert]
Ich war zufrieden und habe die Klasse dann in mein Projekt übernommen und den Code überall entsprechend angepasst.