Kleines Python-Skript, das sich jeden Tag auf Arbeit mit dem Referenz- und Entwicklungs-Datenbankserver verbindet und auf neue Daten überprüft, die von ersteren auf letzteren gespiegelt werden müssen:
import os.path, datetime
conn_str_ziel = "user1/pass1@tnsname1"
conn_str_quelle = "user2/pass2@tnsname2"
query_str = "SELECT MAX(RECHENTERMIN) FROM GROSSE_TABELLE"
msg_titel = "Data Transfer Checker"
msg_head = "Es ist ein neuerer Datenstand vorhanden!"
# aktuelles Datum im ISO-Format (YYYY-MM-DD)
today = datetime.date.today().isoformat()
lastrun = today
lastrun_file = os.path.join(os.path.dirname(os.path.realpath(__file__)),
"datatransfer_check_last.txt")
if os.path.isfile(lastrun_file):
# Durch den with-Context werden Fehler beim Einlesen der Datei abgefangen
# und lastrun behält den Wert von today bei -> keine DB-Anfrage.
with open(lastrun_file, encoding='ascii') as f:
lastrun = f.readline().rstrip()
if lastrun < today:
# Ich verzichte hier auf zusätzliche Wrapper und nutze den Oracle-Treiber
# direkt. Diese funktionieren alle ähnlich dem eingebauten sqlite-Modul.
import cx_Oracle as oracle
rechendatum_ziel = None
rechendatum_quelle = None
# Siehe oben. Bei DB-Fehler ist das Rechendatum unbestimmt.
with oracle.connect(conn_str_ziel) as conn:
cursor = conn.cursor()
cursor.execute(query_str)
result = cursor.fetchone()
if result:
# RECHENTERMIN ist eine DATE-Spalte, die der DB-Treiber als
# datetime.datetime-Objekt zurückgibt. Wir wollen nur den
# Datums-Teil, da ohnehin keine Tageszeit angegeben ist.
rechendatum_ziel = result[0].date().isoformat()
cursor.close()
with oracle.connect(conn_str_quelle) as conn:
cursor = conn.cursor()
cursor.execute(query_str)
result = cursor.fetchone()
if result:
rechendatum_quelle = result[0].date().isoformat()
cursor.close()
if rechendatum_ziel and rechendatum_quelle and rechendatum_quelle > rechendatum_ziel:
msg = "\n".join([msg_head, "Rechendatum Ziel:\t" + rechendatum_ziel,
"Rechendatum Quelle:\t" + rechendatum_quelle])
# ich frickel lieber mit der Win32-API herum als mit Tk :P
import ctypes
MB_OK = 0x00000000
MB_INFORMATION = 0x00000040
# unter Python2 mit Byte-Strings MessageBoxA nutzen!
ctypes.windll.user32.MessageBoxW(None, msg, msg_titel, MB_OK | MB_INFORMATION)
with open(lastrun_file, mode='w', encoding='ascii') as f:
print(today, file=f)
Alles anzeigen
