Hourly backup to HistorianBackup.db

This script automatically creates a backup of historian data by copying records from the main database (Historian.db) to a secondary database (historianBackup.db).

What the script does

• Every time the script runs, it:

  1. Determines the previous full hour in UTC

     • Example: if executed at 14:00 UTC, it processes data from 13:00:00 to 14:00:00

  2. Connects to:

     • The source historian database
     • The backup historian database

  3. Retrieves all records from the previous hour for the following tables:

     • HistoricData
     • HistoricEvents
     • HistoricAlarms
  4. Inserts the retrieved records into the backup database

Key characteristics

• UTC-based operation
  All timestamps are handled in UTC to match the historian database format and avoid timezone issues.
• Safe to run repeatedly
  The script uses INSERT OR IGNORE, ensuring that duplicate records are not inserted if the script runs more than once for the same time window.
• Structure consistency required
  The backup database must have the same table structure as the main historian database.

• Transactional integrity

  All data is copied within a transaction:

  • If successful → changes are committed
  • If an error occurs → changes are rolled back
• Automatic table validation
  The script checks whether each table exists in both databases before attempting to copy data.

Typical use case

This script is typically scheduled to run every hour, providing:

• Continuous backup of historian data
• Protection against data loss
• A secondary database for reporting or redundancy

# Import libraries
import time
from datetime import datetime, timedelta
import sqlite3
import sys
import os

try:
    # Database paths
    source_db = "Database/Historian.db"
    backup_db = "D:/HistorianBackup.db"

    # Make sure source exists
    if not os.path.exists(source_db):
        raise Exception(f"Source database not found: {source_db}")

    # Connect to source and backup databases
    src_con = sqlite3.connect(source_db)
    src_cur = src_con.cursor()

    bak_con = sqlite3.connect(backup_db)
    bak_cur = bak_con.cursor()

    # Determine previous full hour in UTC
    now_utc = datetime.utcnow()
    current_hour_start_utc = now_utc.replace(minute=0, second=0, microsecond=0)
    previous_hour_start_utc = current_hour_start_utc - timedelta(hours=1)
    previous_hour_end_utc = current_hour_start_utc

    timestampfrom = previous_hour_start_utc.strftime('%Y-%m-%d %H:%M:%S')
    timestampto   = previous_hour_end_utc.strftime('%Y-%m-%d %H:%M:%S')

    print(f"Copying historian data in UTC from {timestampfrom} to {timestampto}")

    def copy_table_for_timerange(table_name, time_column):
        # Check if table exists in source
        src_cur.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
            (table_name,)
        )
        if not src_cur.fetchone():
            print(f"Table {table_name} does not exist in source database, skipping.")
            return

        # Check if table exists in backup
        bak_cur.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name=?",
            (table_name,)
        )
        if not bak_cur.fetchone():
            print(f"Table {table_name} does not exist in backup database, skipping.")
            return

        # Get column names from source table
        src_cur.execute(f"PRAGMA table_info({table_name})")
        columns_info = src_cur.fetchall()

        if not columns_info:
            print(f"Could not read columns for table {table_name}, skipping.")
            return

        column_names = [col[1] for col in columns_info]
        column_list = ", ".join(column_names)
        placeholders = ", ".join(["?"] * len(column_names))

        # Read rows from previous UTC hour
        select_query = f"""
            SELECT {column_list}
            FROM {table_name}
            WHERE {time_column} >= ?
              AND {time_column} < ?
            ORDER BY {time_column}
        """

        src_cur.execute(select_query, (timestampfrom, timestampto))
        rows = src_cur.fetchall()

        if not rows:
            print(f"No rows found in {table_name}.")
            return

        # Insert into backup
        insert_query = f"""
            INSERT OR IGNORE INTO {table_name} ({column_list})
            VALUES ({placeholders})
        """

        bak_cur.executemany(insert_query, rows)
        print(f"Copied {len(rows)} row(s) into {table_name}.")

    # Start transaction
    bak_con.execute("BEGIN")

    # Copy historian tables
    copy_table_for_timerange("HistoricData", "Timestamp")
    copy_table_for_timerange("HistoricEvents", "OnTime")
    copy_table_for_timerange("HistoricAlarms", "OnTime")

    # Commit changes
    bak_con.commit()

    # Close connections
    src_con.close()
    bak_con.close()

    print("Historian backup for previous UTC hour completed successfully.")

except Exception as e:
    try:
        bak_con.rollback()
    except:
        pass

    exc_type, exc_obj, exc_tb = sys.exc_info()
    print(exc_type, exc_tb.tb_lineno)
    print(str(e))