Redis - Datenbank Zugriff mit Python

[FrankM]

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.

[FrankM]

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.

[FrankM]

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

• hset
• hexists
• hdel
• hget

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.

Klasse

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.