You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
97 lines
3.6 KiB
97 lines
3.6 KiB
import argparse
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
|
|
# Setup basic configuration for logging
|
|
logging.basicConfig(
|
|
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
|
|
)
|
|
|
|
|
|
def initialize_db(db_path):
|
|
# Check if the database file already exists
|
|
db_exists = os.path.exists(db_path)
|
|
if db_exists:
|
|
logging.info(f"{db_path} exists") # Log a message if the database exists
|
|
return
|
|
try:
|
|
with sqlite3.connect(
|
|
db_path
|
|
) as conn: # Using 'with' to ensure that the connection is closed automatically
|
|
configure_database(conn)
|
|
except sqlite3.Error as e:
|
|
logging.error(f"Database error: {e}") # Log any SQLite errors that occur
|
|
except Exception as e:
|
|
logging.error(
|
|
f"Exception in initialize_db: {e}"
|
|
) # Log any other exceptions that occur
|
|
|
|
|
|
def configure_database(conn):
|
|
cursor = conn.cursor()
|
|
# Setting the journal mode to WAL for better concurrency
|
|
cursor.execute("PRAGMA journal_mode = WAL;")
|
|
# Setting synchronous to NORMAL for a balance between speed and reliability
|
|
cursor.execute("PRAGMA synchronous = NORMAL;")
|
|
# Setting a busy timeout to prevent immediate failures when the database is locked
|
|
cursor.execute("PRAGMA busy_timeout = 5000;")
|
|
# Increasing the cache size to reduce the number of disk I/O operations
|
|
cursor.execute("PRAGMA cache_size = -32000;")
|
|
# Enabling memory-mapped I/O for potentially faster file operations
|
|
cursor.execute("PRAGMA mmap_size = 536870912;")
|
|
# Setting locking mode to EXCLUSIVE can enhance performance for single-user scenarios
|
|
cursor.execute("PRAGMA locking_mode = EXCLUSIVE;")
|
|
# Ensuring foreign key constraints are enforced for data integrity
|
|
cursor.execute("PRAGMA foreign_keys = ON;")
|
|
conn.commit() # Commit all PRAGMA configurations
|
|
|
|
logging.info("Set up database with multi-user optimizations.")
|
|
|
|
|
|
def batch_transact(db_path, operations):
|
|
try:
|
|
with sqlite3.connect(
|
|
db_path
|
|
) as conn: # Ensure that the connection is handled properly
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"BEGIN TRANSACTION;"
|
|
) # Start a transaction for batch operations
|
|
for operation in operations:
|
|
cursor.execute(
|
|
operation
|
|
) # Execute each SQL operation provided in the operations list
|
|
cursor.execute("COMMIT;") # Commit all operations at once
|
|
except sqlite3.Error as e:
|
|
logging.error(f"Database error during batch transaction: {e}")
|
|
except Exception as e:
|
|
logging.error(f"Exception in batch_transact: {e}")
|
|
|
|
|
|
def maintenance(db_path):
|
|
try:
|
|
with sqlite3.connect(db_path) as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"PRAGMA optimize;"
|
|
) # Optimize the database to maintain performance
|
|
cursor.execute("VACUUM;") # Reclaim space and defragment the database file
|
|
except sqlite3.Error as e:
|
|
logging.error(f"Database error during maintenance: {e}")
|
|
except Exception as e:
|
|
logging.error(f"Exception in maintenance: {e}")
|
|
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser(
|
|
description="Initialize and manage an SQLite database."
|
|
)
|
|
parser.add_argument("db_path", type=str, help="Path to the SQLite database file.")
|
|
args = parser.parse_args()
|
|
return args
|
|
|
|
|
|
if __name__ == "__main__":
|
|
args = parse_args() # Parse the command-line arguments for the database path
|
|
initialize_db(args.db_path) # Use the parsed path to initialize the database
|
|
|