Database backup contrib with CmdBackup

evennia/contrib/utils/database_backup/README.md

@@ -0,0 +1,69 @@
+# Database Backup Scheduler
+
+Contribution by helpme (2024)
+
+This module helps backup and restore your game world from database, as well as scheduling backups in-game. Backups are saved in your game's `server` folder. Database backups are *not* automatically uploaded to any cloud service, it is left to you to decide what to do with them (i.e. pushed to git, uploaded to a cloud service, downloaded to hard drive).
+
+Backups can take place at any time, while restoring the game world from backup has to take place outside of game commands, during downtime, as documented below.
+
+Currently, the sqlite3 (the evennia default) and postgresql databases are supported. Others are welcome to add more.
+
+## Installation
+
+This utility adds the `backup` command. Import the module into your commands and add it to your command set to make it available.
+
+In `mygame/commands/default_cmdsets.py`:
+
+```python
+...
+from evennia.contrib.utils.database_backup import DbCmdSet   # <---
+
+class CharacterCmdset(default_cmds.Character_CmdSet):
+    ...
+    def at_cmdset_creation(self):
+        ...
+        self.add(DbCmdSet)  # <---
+
+```
+
+Then `reload` to make the `backup` command available.
+
+## Permissions
+
+By default, the backup command is only available to those with Developer permissions and higher. You can change this by overriding the command and setting its locks from "cmd:pperm(Developer)" to the lock of your choice.
+
+## Settings Used
+
+This utility uses the settings.DATABASES dictionary.
+
+## Restoration
+
+Remember to `evennia stop` before restoring your db.
+
+### Restoring sqlite3 (.db3)
+
+* Copy the database backup you want to restore from back into the `server/` directory
+* Rename the database backup to `evennia.db3` if you have not modified `settings.DATABASES`, otherwise whatever name is in your `settings.DATABASES` dictionary.
+
+### Restoring postgres (.sql)
+
+* Prepare the following variables
+```
+export DB_USER=db_user from your settings.DATABASES
+export DB_NAME=db_name from your settings.DATABASES
+export BACKUP_FILE=the path to the backup file you are restoring from
+export PGPASSWORD=your db password
+
+If you prefer not to export your password to an env variable, you can enter it when prompted instead.
+```
+* Run the following commands
+```
+# Drop the existing database if it exists
+psql -U $DB_USER -c "DROP DATABASE IF EXISTS $DB_NAME;" || exit 1
+
+# Recreate the database
+psql -U $DB_USER -c "CREATE DATABASE $DB_NAME;" || exit 1
+
+# Restore the database from the backup file
+psql -U $DB_USER -d $DB_NAME -f $BACKUP_FILE || exit 1
+```

evennia/contrib/utils/database_backup/__init__.py

@@ -0,0 +1,5 @@
+"""
+Database backups - helpme 2024
+"""
+
+from .database_backup import DbCmdSet  # noqa

evennia/contrib/utils/database_backup/database_backup.py

@@ -0,0 +1,183 @@
+from django.conf import settings
+
+from evennia.comms.models import ChannelDB
+from evennia import CmdSet, DefaultScript
+from evennia.commands.default.muxcommand import MuxCommand
+from evennia.utils.create import create_script
+from evennia.utils import logger, search
+
+import datetime
+import os
+import shutil
+import subprocess
+
+_MUDINFO_CHANNEL = None
+BACKUP_FOLDER = "server/backups"
+DATETIME_FORMAT_STR = "%Y-%m-%d.%H_%M_%S"
+DEFAULT_INTERVAL = 86400
+
+
+class DatabaseBackupScript(DefaultScript):
+    """
+    The global script to backup the server on a schedule.
+
+    It will be automatically created the first time the `backup` command is used.
+    """
+
+    def at_script_creation(self):
+        super().at_script_creation()
+        self.key = "db_backup_script"
+        self.desc = "Database backups"
+        self.persistent = True
+
+    def backup_postgres(self, db_name, db_user, output_file_path):
+        """
+        Run `pg_dump` on the postgreSQL database and save the output.
+
+        Returns:
+            str: Success message
+        """
+
+        output_file_path += ".sql"
+        subprocess.run(
+            ["pg_dump", "-U", db_user, "-F", "p", db_name],
+            stdout=open(output_file_path, "w"),
+            check=True,
+        )
+        return f"|wPostgreSQL db backed up in: {BACKUP_FOLDER}|n"
+
+    def backup_sqlite3(self, db_name, output_file_path):
+        """
+        Copy the sqlite3 db.
+
+        Returns:
+            str: Success message
+        """
+
+        output_file_path += ".db3"
+        os.makedirs(os.path.dirname(output_file_path), exist_ok=True)
+        shutil.copy(db_name, output_file_path)
+        return f"|wsqlite3 db backed up in {BACKUP_FOLDER}|n"
+
+    def at_repeat(self):
+        global _MUDINFO_CHANNEL
+        if not _MUDINFO_CHANNEL:
+            if settings.CHANNEL_MUDINFO:
+                _MUDINFO_CHANNEL = ChannelDB.objects.get(db_key=settings.CHANNEL_MUDINFO["key"])
+
+        databases = settings.DATABASES
+        db = databases["default"]
+        engine = db.get("ENGINE")
+        db_name = db.get("NAME")
+        db_user = db.get("USER")
+
+        try:
+            # Create the output folder if it doesn't exist
+            os.makedirs(BACKUP_FOLDER, exist_ok=True)
+            output_file = datetime.datetime.now().replace(microsecond=0).isoformat()
+            output_file_path = os.path.join(BACKUP_FOLDER, output_file)
+
+            if "postgres" in engine:
+                message = self.backup_postgres(db_name, db_user, output_file_path)
+            elif "sqlite3" in engine:
+                message = self.backup_sqlite3(db_name, output_file_path)
+
+            logger.log_sec(message)
+            if _MUDINFO_CHANNEL:
+                _MUDINFO_CHANNEL.msg(message)
+        except Exception as e:
+            logger.log_err("Backup failed: {}".format(e))
+
+
+class CmdBackup(MuxCommand):
+    """
+    Backup your database to the server/backups folder in your game directory.
+
+    Usage:
+      backup [interval in seconds] - Schedule a backup. The default interval is one day.
+      backup/stop  - Stop the backup script (equivalent to scripts/delete #id)
+      backup/force - Trigger a backup manually
+    """
+
+    key = "backup"
+    aliases = "backups"
+    locks = "cmd:pperm(Developer)"
+
+    def get_latest_backup(self):
+        try:
+            files = os.listdir(BACKUP_FOLDER)
+            paths = [os.path.join(BACKUP_FOLDER, basename) for basename in files]
+            last_backup = max(paths, key=os.path.getctime)
+            return last_backup
+        except Exception:
+            return ""
+
+    def create_script(self, interval):
+        """Create new script. Deletes old script if it exists."""
+        script = search.search_script("db_backup_script")
+        if script:
+            script[0].delete()
+        create_script(DatabaseBackupScript, interval=interval)
+        self.caller.msg(f"You have scheduled backups to run every {interval} seconds.")
+
+    def get_script(self):
+        """
+        Returns:
+            script: Existing script
+        """
+        script = search.search_script("db_backup_script")
+        if script:
+            return script[0]
+
+    def func(self):
+        """
+        Database backup functionality
+        """
+        caller = self.caller
+        args = self.args.strip()
+        interval = int(args) if args.isnumeric() else DEFAULT_INTERVAL
+
+        script = self.get_script()
+
+        # Kill existing backup script
+        if "stop" in self.switches:
+            if not script:
+                caller.msg("No existing db backup script to delete.")
+                return
+            script.delete()
+            caller.msg("DB backup script deleted.")
+            return
+
+        # Create new backup script
+        if not script:
+            self.create_script(interval)
+            return
+
+        # Change backup script's interval (if the provided interval is different)
+        original_interval = script.interval
+        if args and original_interval != interval:
+            self.create_script(interval)
+            return
+
+        if "force" in self.switches:
+            # Manually trigger the backup
+            script.at_repeat()
+            return
+
+        if self.get_latest_backup():
+            caller.msg(f"Most recent database backup: {self.get_latest_backup()}.")
+        else:
+            caller.msg(f"No database backups found in {BACKUP_FOLDER}")
+
+        caller.msg(
+            f"Countdown till next scheduled backup: |x{datetime.timedelta(seconds=script.time_until_next_repeat())}|n. Use |wbackup/force|n to manually backup the database."
+        )
+
+
+class DbCmdSet(CmdSet):
+    """
+    Database backup command
+    """
+
+    def at_cmdset_creation(self):
+        self.add(CmdBackup)