Skip to content

Redis - Datenbank Zugriff mit Python

Redis
  • 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.

    87922ef6-24c0-492f-be67-a97944f79e0c-grafik.png

    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.

  • 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

    • 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.

    393195f7-1017-4285-8fca-734ee6b4bff7-grafik.png

    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.

  • Pycharm - Umzug auf neuen Rechner

    Linux
    1
    0 Stimmen
    1 Beiträge
    87 Aufrufe
    Niemand hat geantwortet
  • Example Class

    Angeheftet Python3
    3
    0 Stimmen
    3 Beiträge
    140 Aufrufe
    FrankMF

    In meinem PywebIO Projekt tauchte heute ein alter Bekannter auf. Wenn ich einen Eintrag (innerhalb der Liste) löschte, war das etwas durcheinander 🤔

    Ja, den Fehler kenne ich schon was länger und stolper immer mal wieder drüber. Heute z.B.

    Also mal grübeln 🤓

    Das mache ich, wenn ich einen Eintrag lösche

    case 'Yes': # delete entrie BackupList.delete((line - 1)) # save json BackupList.save_json() # Load data from filesystem BackupList.load_json() # Reload Tab Backup backup()

    Ich mache folgendes

    lösche den Eintrag speicher die Liste lade die Liste baue den Tab wieder auf

    Das führt aber dazu, das meine Liste die als Objekt im Speicher steht nicht aktuell ist. Im Gegenteil, da ist dann etwas Unordnung. Ich brauchte also die Möglichkeit mein Objekt backups irgendwie zurückzusetzen.

    Mal gegoogelt und die Dinge sind manchmal wirklich total easy 🙂

    # clear dict backups.clear()

    Das leert das Objekt und im nächsten Schritt kann ich es wieder befüllen.

    Lösung case 'Yes': # delete entrie BackupList.delete((line - 1)) # save json BackupList.save_json() # clear dict backups.clear() # Load data from filesystem BackupList.load_json() # Reload Tab Backup backup()

    Problem erledigt. Damit ich das noch finde, wenn mein Kopf das nicht mehr hergibt, notiere ich das hier.

  • Python - Dict -> JSON und umgekehrt

    Python3
    1
    0 Stimmen
    1 Beiträge
    299 Aufrufe
    Niemand hat geantwortet
  • ResticUI with PyWebIO - Repo

    PyWebIO
    1
    0 Stimmen
    1 Beiträge
    94 Aufrufe
    Niemand hat geantwortet
  • Python3 - VSCodium Language Settings

    Python3
    3
    0 Stimmen
    3 Beiträge
    141 Aufrufe
    FrankMF

    Ich habe da jetzt gestern stundenlang mit rum gespielt, insbesondere Deinstallation, Neuinstallation, VSCode Installation usw. Das ging einfach nicht.

    Irgendwo meine ich dann gelesen zu haben, das VSCode auf python 3.9 festgenagelt ist und deswegen so einige Dinge nicht gehen.

    Nagelt mich darauf nicht fest, ich bin in dieser Coding Welt kein Profi.

    Dann fiel mir ein, das ich schon vor einigen Monaten mal kurz einen Blick in Pycharm geworfen hatte. Gefiel mir damals gar nicht.

    Das habe ich heute mal auf die Platte geworfen um zu schauen, ob das vernünftig mit dem o.g. Problem umgehen kann.

    6f1c3906-05a3-4126-81d4-7fdf6aa27faa-grafik.png

    Puuuh, da kann man ja noch viel mehr einstellen als bei VSCode, oder mindestens ähnlich viel. Das kann einen ganz schön überfordern. Also Stück für Stück die nächsten Tage.

    Was ist mir positiv aufgefallen?

    Wenn ich VSCodium neu installiert habe, habe ich unheimlich mit Python Versionen, pipenv usw. zu kämpfen gehabt. Bis da mein Projekt wieder lief, konnte schon mal was Zeit vergehen.

    Mit Pycharm Ordner ausgewählt, angeklickt das man dem Coder vertraut. Noch eine Python Version ausgewählt, danach fragt er das er eine pipenv Umgebung gefunden hat und diese nutzen möchte. Ja angeklickt und fertig. Keine zwei Minuten und das Projekt lief wieder 🤓

    Erster Git Commit ging auch sofort und ohne Murren.

    Schon mal ganz viele Pluspunkte gesammelt.

    Was mich aktuell beschäftigt, ist der Tab Problems, da hatte ich vorher deutlich weniger. Da muss ich mich noch was einarbeiten und schauen, wo man das alles einstellen kann.

    Ich halte euch auf dem Laufenden..

    Vielleicht fliegt ja das VSCo** Zeug runter!? Mal sehen..

  • Redis Replication

    Angeheftet Verschoben Redis
    4
    1 Stimmen
    4 Beiträge
    458 Aufrufe
    FrankMF

    Um die Verbindung zu testen, kann man folgende Befehle nutzen.

    redis-cli -h 10.1.1.0 -p 6379 -a <PASSWORD>

    und

    telnet 10.1.1.0 6379
  • Redis installieren

    Angeheftet Verschoben Redis
    1
    0 Stimmen
    1 Beiträge
    383 Aufrufe
    Niemand hat geantwortet
  • Redis Datenbank sichern

    Verschoben Redis
    1
    0 Stimmen
    1 Beiträge
    763 Aufrufe
    Niemand hat geantwortet