# Datei dbhandler.py import mysql.connector import datetime import configparser import logging from logging import config import json class DBHandler(logging.Handler): def __init__(self): logging.Handler.__init__(self) # Datenbank Konfiguration lesen confparser = configparser.ConfigParser() confparser.read('db.ini') # Verbindung zur DB herstellen self.connection= mysql.connector.connect(host=confparser['pidb']['host'],user = confparser['pidb']['user'],password = confparser['pidb']['password'],database = confparser['pidb']['database']) # Tabelle anlegen, falls noch keine existiert. self.connection.cursor().execute('''CREATE TABLE IF NOT EXISTS logs ( id INTEGER PRIMARY KEY AUTO_INCREMENT, zeit VARCHAR(50), level VARCHAR(8), message VARCHAR(40), logger_name VARCHAR(25), filename VARCHAR(60), line INTEGER)''') self.connection.commit() # Bei Löschung der Klasseninstanz, Connection schliessen def __del__(self): if self.connection != None: self.connection.close() self.connection = None # LogRecord in die DB schreiben def emit(self, record): try: # Log-Daten in ein Tupel extrahieren log_entry = (datetime.datetime.utcnow().isoformat(), record.levelname, record.getMessage(), record.name, record.pathname, record.lineno) # Log in die Datenbank einfügen cursor = self.connection.cursor() cursor.execute('''INSERT INTO logs (zeit, level, message, logger_name, filename, line) VALUES (%s, %s, %s, %s, %s, %s)''', log_entry) self.connection.commit() cursor.close() except Exception as e: # im Fehlerfall den LogRecord nach "oben" weiterreichen. self.handleError(record) # Config aus JSON Datei laden und setzen with open('handler_config.json') as file_config: config.dictConfig(json.load(file_config)) # Logging testen def testlogging(): print('testlogging() aufgerufen.') logging.debug('Diese Zeile wird ausgefiltert.') logging.debug('DEBUG/Druckdaten werden gesendet.') logging.info('INFO/Es werden 2 Seiten gedruckt') logging.debug('Diese Zeile wird auch ausgefiltert.') logging.warning('WARN/nicht genügend Papier im Drucker') logging.error('ERROR/Drucker nicht gefunden.') logging.critical('CRITICAL/Drucker brennt') if __name__ == '__main__': testlogging()