sqlBackup: v0.1

Author: klevze

.gitignore

@@ -171,4 +171,4 @@ cython_debug/
 .pypirc
 
 .vscode
-config.ini
+config.ini

README.md

@@ -28,6 +28,9 @@
 - **Remote Uploads:**  
   Optionally upload backups to a remote server using SFTP, FTP, or SCP with configurable scheduling (daily, first day, last day, specific weekday, or a numeric day of the month).
 
+- **Wildcard Support for Ignored Databases:**  
+  Use wildcard patterns (e.g., `projekti_*`) in `ignored_databases` to skip databases by name pattern.
+
 - **Modular & Maintainable:**  
   Code is organized into multiple modules (configuration, backup logic, notifications, remote upload) for easier maintenance and extensibility.
 
@@ -49,14 +52,16 @@
 
 ### Prerequisites
 
-- **Python 3.6+** is required.
-- Install the necessary Python packages using pip:
-  ```bash
-  pip install requests paramiko twilio
-  ```
-  - `requests`: For HTTP requests (used in notifications and remote uploads).
-  - `paramiko`: For SFTP uploads.
-  - `twilio`: For sending SMS notifications.
+1. **Python 3.6+** is required.
+2. **MySQL or MariaDB** client tools installed (e.g., `mysql`, `mysqldump`).
+3. Install the necessary Python packages using `pip`:
+   ```bash
+   pip install requests paramiko twilio fnmatch 
+   ```
+   - `requests`: For HTTP requests (used in notifications and remote uploads).
+   - `paramiko`: For SFTP uploads.
+   - `twilio`: For sending SMS notifications.
+   - `fnmatch`: For wildcard support
 
 ### Setup
 
@@ -80,13 +85,21 @@
        └── remote_upload.py  # Remote upload functionality
    ```
 
-3. **Configure the Project:**
-   - **Important:** Do not modify `config.ini.default` directly. Instead, copy it to create your local configuration file:
+3. **Configuration File Setup:**
+   - **Important:** Do not modify `config.ini.default` directly. Instead, copy it:
      ```bash
      cp config.ini.default config.ini
      ```
    - Open `config.ini` and adjust the settings to match your environment (e.g., MySQL credentials, notification channel settings, remote upload settings, etc.).
 
+   > **Note:** If you pull new changes from this repo in the future, your local `config.ini` will remain untouched, preserving your production settings.
+
+4. **(Optional) Unit Tests:**
+   - If you want to run the included unit tests (if any), install `unittest` (bundled with Python), plus any additional test dependencies, and run:
+     ```bash
+     python3 -m unittest discover
+     ```
+
 ## Configuration Tutorial
 
 The `config.ini` file is the central configuration file for **sqlBackup**. It is divided into several sections:
@@ -101,6 +114,7 @@ The `config.ini` file is the central configuration file for **sqlBackup**. It is
 - **mysql_path:** Path to the MySQL client.
 - **mysqldump_path:** Path to the mysqldump utility.
 - **ignored_databases:** Comma-separated list of databases to skip.
+  - **Now supports wildcards:** e.g. `sys, mysql, projekti_*`. Any database name matching `projekti_*` will be ignored (e.g., `projekti_alpha`, `projekti_1`).
 
 ### [telegram]
 - **enabled:** Enable or disable Telegram notifications.
@@ -142,7 +156,7 @@ The `config.ini` file is the central configuration file for **sqlBackup**. It is
 ### [export]
 - **include_routines:** Include stored procedures and functions.
 - **include_events:** Include scheduled events.
-- **column_statistics:** If set to false, the script will add `--column-statistics=0` to the dump command.
+- **column_statistics:** If set to false, the script adds `--column-statistics=0` to the dump command (helpful for older servers).
 
 ### [remote]
 - **upload_enabled:** Enable or disable remote upload of backups.
@@ -162,7 +176,7 @@ python3 main.py
 ```
 
 The script will:
-- Connect to MySQL and dump databases (skipping those in `ignored_databases`).
+- Connect to MySQL and dump databases (skipping those in `ignored_databases`, including wildcards).
 - Archive each dump according to the specified format.
 - Display a summary table with database name, backup status, elapsed time, dump size, and archive size.
 - Send notifications via the enabled channels.

config.ini

@@ -13,7 +13,7 @@ from src.remote_upload import upload_backups
 
 def main():
     config = load_config()
-    
+        
     # Run backups and get summary.
     errors, summary = run_backups(config)
     if errors:

sqlBackup

@@ -7,6 +7,7 @@
 import time
 import threading
 from itertools import cycle
+from fnmatch import fnmatch
 
 from .config import CONFIG
 
@@ -20,7 +21,8 @@
 # --- Configuration Variables ---
 BACKUP_DIR = CONFIG.get("backup", "backup_dir")
 ARCHIVE_FORMAT = CONFIG.get("backup", "archive_format").lower()
-IGNORED_DATABASES = set(x.strip() for x in CONFIG.get("mysql", "ignored_databases").split(','))
+# Split, then strip whitespace from each pattern
+IGNORED_DB_PATTERNS = [p.strip() for p in CONFIG.get("mysql", "ignored_databases").split(",")]
 MYSQL = CONFIG.get("mysql", "mysql_path")
 MYSQLDUMP = CONFIG.get("mysql", "mysqldump_path")
 TIMESTAMP = datetime.datetime.now().strftime("%F")
@@ -63,7 +65,7 @@ def format_size(size: int) -> str:
         return f"{size / (1024 * 1024 * 1024):.1f} GB"
 
 def create_temp_mysql_config() -> str:
-    """Create a temporary MySQL client config file from the [mysql] section."""
+    from .config import CONFIG
     tmp = tempfile.NamedTemporaryFile(mode="w", delete=False)
     tmp.write("[client]\n")
     tmp.write(f"user = {CONFIG.get('mysql', 'user')}\n")
@@ -104,6 +106,16 @@ def get_all_databases() -> list:
         print(f"{RED}Error retrieving databases: {e.stderr}{RESET}")
         sys.exit(1)
 
+def is_ignored(db_name: str) -> bool:
+    """
+    Return True if db_name matches any of the patterns in IGNORED_DB_PATTERNS.
+    We use fnmatch to support wildcards (like projekti_*).
+    """
+    for pattern in IGNORED_DB_PATTERNS:
+        if fnmatch(db_name, pattern):
+            return True
+    return False
+
 def backup_database(db: str) -> tuple:
     """
     Dump the given database, archive it, and return a tuple:
@@ -248,18 +260,22 @@ def run_backups(config) -> tuple:
     databases = get_all_databases()
     errors = []
     summary_lines = []
+    
     print_table_header()
     for db in databases:
-        if db in IGNORED_DATABASES:
+        if is_ignored(db):
             print_table_row(db, "Skipped", "-", "-", "-")
             continue
+
         start = time.time()
         status, dump_size, archive_size = backup_database(db)
         elapsed = f"{time.time() - start:.1f}"
         if status == "Error":
             errors.append(db)
+        
         print_table_row(db, status, elapsed, dump_size, archive_size)
         summary_lines.append(f"{db}: {status} in {elapsed}s")
+    
     separator = f"|{'-'*27}|{'-'*17}|{'-'*12}|{'-'*14}|{'-'*16}|"
     print(separator)
     summary = "\n".join(summary_lines)

src/backup.py

@@ -48,19 +48,20 @@ def send_email_notification(config, message: str) -> None:
         except Exception as e:
             print(f"{RED}Email notification failed: {e}{RESET}")
 
-def send_slack_notification(config, message: str) -> None:
+def send_slack_notification(config, message):
     if config.has_section("slack") and config.getboolean("slack", "enabled", fallback=False):
         webhook_url = config.get("slack", "webhook_url")
         try:
             payload = {"text": message}
+            # Note: use json=payload, not data=payload
             response = requests.post(webhook_url, json=payload)
             if response.status_code != 200:
-                print(f"{RED}Slack notification failed: {response.text}{RESET}")
+                print(f"Slack notification failed: {response.text}")
             else:
-                print(f"{BLUE}Slack notification sent.{RESET}")
+                print("Slack notification sent.")
         except Exception as e:
-            print(f"{RED}Slack notification error: {e}{RESET}")
-
+            print(f"Slack notification error: {e}")
+            
 def send_sms_notification(config, message: str) -> None:
     if config.has_section("sms") and config.getboolean("sms", "enabled", fallback=False):
         try:

src/notifications.py

@@ -0,0 +1 @@
+# __init__.py (empty)

tests/__init__.py

@@ -0,0 +1,26 @@
+import unittest
+import datetime
+from src.backup import format_size, should_upload
+
+class TestBackupHelpers(unittest.TestCase):
+    def test_format_size_bytes(self):
+        self.assertEqual(format_size(500), "500 B")
+
+    def test_format_size_kb(self):
+        self.assertEqual(format_size(2048), "2.0 KB")
+
+    def test_format_size_mb(self):
+        self.assertEqual(format_size(1048576), "1.0 MB")
+
+    def test_format_size_gb(self):
+        self.assertEqual(format_size(1073741824), "1.0 GB")
+
+    def test_should_upload_daily(self):
+        self.assertTrue(should_upload("daily"))
+
+    def test_should_upload_numeric(self):
+        now = datetime.datetime.now()
+        self.assertTrue(should_upload(str(now.day)))
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_backup.py

@@ -0,0 +1,13 @@
+import unittest
+from src.config import load_config
+
+class TestConfig(unittest.TestCase):
+    def test_config_loaded(self):
+        config = load_config()
+        # Check that required sections exist.
+        self.assertTrue(config.has_section("backup"))
+        self.assertTrue(config.has_section("mysql"))
+        self.assertTrue(config.has_section("notification"))
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_config.py

@@ -0,0 +1,61 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from src.notifications import send_telegram_notification, send_email_notification, send_slack_notification
+from src.config import CONFIG
+
+class TestNotifications(unittest.TestCase):
+    @patch("src.notifications.requests.post")
+    def test_send_telegram_notification(self, mock_post):
+        # Prepare a dummy config with telegram enabled
+        dummy_config = CONFIG
+        dummy_config["telegram"] = {
+            "enabled": "true",
+            "telegram_token": "dummy_token",
+            "telegram_chatid": "dummy_chat"
+        }
+        send_telegram_notification(dummy_config, "Test Telegram message")
+        self.assertTrue(mock_post.called, "requests.post was not called for Telegram")
+
+    @patch("src.notifications.smtplib.SMTP")
+    def test_send_email_notification(self, mock_smtp):
+        dummy_config = CONFIG
+        dummy_config["email"] = {
+            "enabled": "true",
+            "smtp_server": "smtp.example.com",
+            "smtp_port": "587",
+            "username": "user@example.com",
+            "password": "secret",
+            "from_address": "from@example.com",
+            "to_addresses": "to@example.com"
+        }
+        send_email_notification(dummy_config, "Test Email message")
+        self.assertTrue(mock_smtp.called, "SMTP was not called for Email")
+
+    @patch("src.notifications.requests.post")
+    def test_send_slack_notification(self, mock_post):
+        # Configure the mock to simulate a successful Slack response
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.text = "OK"
+        mock_post.return_value = mock_response
+
+        dummy_config = CONFIG
+        dummy_config["slack"] = {
+            "enabled": "true",
+            "webhook_url": "https://hooks.slack.com/services/dummy"
+        }
+
+        # Call the Slack notification function
+        send_slack_notification(dummy_config, "Test Slack message")
+
+        # Ensure requests.post was actually called
+        self.assertTrue(mock_post.called, "requests.post was not called for Slack")
+
+        # Verify we call requests.post with the correct URL and JSON payload
+        mock_post.assert_called_once_with(
+            "https://hooks.slack.com/services/dummy",
+            json={"text": "Test Slack message"}
+        )
+
+if __name__ == "__main__":
+    unittest.main()