mirror of
https://github.com/evennia/evennia.git
synced 2026-08-18 11:14:44 +00:00
Merge pull request #3644 from Machine-Garden-MUD/database_backup_contrib
Database backup contrib to schedule database backups into /server/backups
This commit is contained in:
Griatch
GitHub
84
evennia/contrib/utils/database_backup/README.md
Normal file
84
evennia/contrib/utils/database_backup/README.md
Normal file
@ -0,0 +1,84 @@
|
||||
# Database Backup Scheduler
|
||||
|
||||
Contribution by helpme (2024)
|
||||
|
||||
This module schedules backups in-game, which saves a copy of your database to your game's `server/backups` 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 the ether, downloaded to a hard drive).
|
||||
|
||||
Backups can take place at any time. Restoring the game world from backup takes place 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. The `backup` command can be used to set up a scheduled backup script, or trigger the script to run immediately. The backup script makes a backup of your game world. 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.
|
||||
|
||||
If you prefer to run the script without the `backup` command, you can manage it as a global script in your settings:
|
||||
|
||||
```python
|
||||
# in mygame/server/conf/settings.py
|
||||
|
||||
GLOBAL_SCRIPTS = {
|
||||
"backupscript": {
|
||||
"typeclass": "evennia.contrib.utils.database_backup.DatabaseBackupScript",
|
||||
"repeats": -1,
|
||||
"interval": 86400,
|
||||
"desc": "Database backup script"
|
||||
},
|
||||
}
|
||||
```
|
||||
|
||||
## 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
|
||||
* By default (unless you changed the name of the file in `settings DATABASES`), the game data is expected to be located in the sqlite3 file `mygame/server/evennia.db3`. Copy your backup file over this file to recover your backup.
|
||||
|
||||
### Restoring postgres (.sql)
|
||||
|
||||
* Prepare the following variables
|
||||
```
|
||||
export DB_USER=db_user # db_user from your settings.DATABASES
|
||||
export DB_NAME=db_name # db_name from your settings.DATABASES
|
||||
export BACKUP_FILE=backup_file_path # the path to the backup file you are restoring from
|
||||
export PGPASSWORD=db_password # the password to your db
|
||||
|
||||
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
|
||||
```
|
||||
5
evennia/contrib/utils/database_backup/__init__.py
Normal file
5
evennia/contrib/utils/database_backup/__init__.py
Normal file
@ -0,0 +1,5 @@
|
||||
"""
|
||||
Database backups - helpme 2024
|
||||
"""
|
||||
|
||||
from .database_backup import DbCmdSet, DatabaseBackupScript # noqa
|
||||
234
evennia/contrib/utils/database_backup/database_backup.py
Normal file
234
evennia/contrib/utils/database_backup/database_backup.py
Normal file
@ -0,0 +1,234 @@
|
||||
r"""
|
||||
Database Backup - helpme 2024
|
||||
|
||||
This adds a `backup` command which can be used to schedule automatic database backups,
|
||||
or trigger one immediately. Backups are saved to your game's `server/backups` folder.
|
||||
|
||||
Currently supports sqlite3 (the Evennia default) and PostgreSQL. SQLite backups use
|
||||
Python's built-in sqlite3 backup API, which is safe to run against a live database.
|
||||
|
||||
See the README for full installation instructions, settings, and restoration steps.
|
||||
"""
|
||||
|
||||
from django.conf import settings
|
||||
|
||||
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 subprocess
|
||||
import sqlite3
|
||||
|
||||
_MUDINFO_CHANNEL = None
|
||||
BACKUP_FOLDER = os.path.join(settings.GAME_DIR, "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 log(self, message):
|
||||
global _MUDINFO_CHANNEL
|
||||
if not _MUDINFO_CHANNEL and settings.CHANNEL_MUDINFO:
|
||||
channels = search.search_channel(settings.CHANNEL_MUDINFO["key"])
|
||||
if channels:
|
||||
_MUDINFO_CHANNEL = channels[0]
|
||||
|
||||
if _MUDINFO_CHANNEL:
|
||||
_MUDINFO_CHANNEL.msg(message)
|
||||
logger.log_sec(message)
|
||||
|
||||
def backup_postgres(self, db_name, db_user, output_file_path):
|
||||
"""
|
||||
Run `pg_dump` on the postgreSQL database and save the output.
|
||||
"""
|
||||
output_file_path += ".sql"
|
||||
try:
|
||||
with open(output_file_path, "w") as output_file:
|
||||
subprocess.run(
|
||||
["pg_dump", "-U", db_user, "-F", "p", db_name],
|
||||
stdout=output_file,
|
||||
check=True,
|
||||
)
|
||||
except (subprocess.CalledProcessError, FileNotFoundError) as e:
|
||||
if os.path.exists(output_file_path):
|
||||
os.remove(output_file_path)
|
||||
raise
|
||||
self.log(f"|wpostgresql db backed up in: {BACKUP_FOLDER}|n")
|
||||
|
||||
def backup_sqlite3(self, db_name, output_file_path):
|
||||
"""
|
||||
Copy the sqlite3 db.
|
||||
"""
|
||||
|
||||
output_file_path += ".db3"
|
||||
os.makedirs(os.path.dirname(output_file_path), exist_ok=True)
|
||||
|
||||
src = sqlite3.connect(db_name)
|
||||
dst = sqlite3.connect(output_file_path)
|
||||
try:
|
||||
src.backup(dst)
|
||||
except sqlite3.DatabaseError:
|
||||
dst.close()
|
||||
src.close()
|
||||
try:
|
||||
os.remove(output_file_path)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
self.log(f"|rsqlite3 db backup to {BACKUP_FOLDER} failed|n")
|
||||
return
|
||||
finally:
|
||||
dst.close()
|
||||
src.close()
|
||||
|
||||
self.log(f"|wsqlite3 db backed up in: {BACKUP_FOLDER}|n")
|
||||
|
||||
def at_repeat(self):
|
||||
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().strftime(DATETIME_FORMAT_STR)
|
||||
output_file_path = os.path.join(BACKUP_FOLDER, output_file)
|
||||
|
||||
if "postgres" in engine:
|
||||
self.backup_postgres(db_name, db_user, output_file_path)
|
||||
elif "sqlite3" in engine:
|
||||
self.backup_sqlite3(db_name, output_file_path)
|
||||
|
||||
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
|
||||
if basename.endswith((".db3", ".sql"))
|
||||
]
|
||||
if not paths:
|
||||
return ""
|
||||
return max(paths, key=os.path.getctime)
|
||||
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 at_pre_cmd(self):
|
||||
databases = settings.DATABASES
|
||||
db = databases["default"]
|
||||
engine = db.get("ENGINE")
|
||||
|
||||
if "postgres" not in engine and "sqlite3" not in engine:
|
||||
self.caller.msg(
|
||||
f"|rDatabase backup failed: unsupported engine '{engine}'. Contrib supports postgres and sqlite3.|n"
|
||||
)
|
||||
return True
|
||||
return super().at_pre_cmd()
|
||||
|
||||
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
|
||||
|
||||
# Manually trigger the backup
|
||||
if "force" in self.switches:
|
||||
if not script:
|
||||
self.create_script(interval)
|
||||
script = self.get_script()
|
||||
script.at_repeat()
|
||||
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
|
||||
|
||||
latest_backup = self.get_latest_backup()
|
||||
if latest_backup:
|
||||
caller.msg(f"Most recent database backup: {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)
|
||||
181
evennia/contrib/utils/database_backup/tests.py
Normal file
181
evennia/contrib/utils/database_backup/tests.py
Normal file
@ -0,0 +1,181 @@
|
||||
"""
|
||||
Tests for database backups.
|
||||
"""
|
||||
|
||||
import evennia.contrib.utils.database_backup.database_backup as backup
|
||||
from evennia.commands.default.tests import BaseEvenniaCommandTest
|
||||
from unittest.mock import patch
|
||||
from evennia.utils import search
|
||||
import subprocess, os, tempfile, sqlite3
|
||||
|
||||
EXCEPTION_STR = "failed"
|
||||
|
||||
|
||||
class TestDatabaseBackupScript(BaseEvenniaCommandTest):
|
||||
mocked_db_setting_postgres = patch(
|
||||
"django.conf.settings.DATABASES",
|
||||
{
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.postgresql",
|
||||
"NAME": "fake_name",
|
||||
"USER": "fake_user",
|
||||
}
|
||||
},
|
||||
)
|
||||
mocked_db_setting_sqlite = patch(
|
||||
"django.conf.settings.DATABASES",
|
||||
{
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.sqlite3",
|
||||
"NAME": "fake_name",
|
||||
}
|
||||
},
|
||||
)
|
||||
mocked_db_setting_unsupported = patch(
|
||||
"django.conf.settings.DATABASES",
|
||||
{
|
||||
"default": {
|
||||
"ENGINE": "django.db.backends.oracle",
|
||||
"NAME": "fake_name",
|
||||
"USER": "fake_user",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self._tmp_dir = tempfile.TemporaryDirectory()
|
||||
self._backup_folder_patch = patch.object(backup, "BACKUP_FOLDER", self._tmp_dir.name)
|
||||
self._backup_folder_patch.start()
|
||||
|
||||
def tearDown(self):
|
||||
super().tearDown()
|
||||
script = search.search_script("db_backup_script")
|
||||
if script:
|
||||
script[0].delete()
|
||||
self._backup_folder_patch.stop()
|
||||
self._tmp_dir.cleanup()
|
||||
|
||||
@mocked_db_setting_sqlite
|
||||
@patch("sqlite3.connect")
|
||||
@patch("evennia.utils.logger.log_sec")
|
||||
def test_sqlite_success(self, mock_logger, mock_connect):
|
||||
self.call(
|
||||
backup.CmdBackup(),
|
||||
"300",
|
||||
"You have scheduled backups to run every 300 seconds.",
|
||||
caller=self.char1,
|
||||
)
|
||||
self.call(backup.CmdBackup(), "/force", "", caller=self.char1)
|
||||
mock_logger.assert_called_with(f"|wsqlite3 db backed up in: {backup.BACKUP_FOLDER}|n")
|
||||
self.call(
|
||||
backup.CmdBackup(),
|
||||
"/stop",
|
||||
"DB backup script deleted.",
|
||||
caller=self.char1,
|
||||
)
|
||||
|
||||
@mocked_db_setting_sqlite
|
||||
@patch("sqlite3.connect")
|
||||
@patch("evennia.utils.logger.log_err")
|
||||
def test_sqlite_failure(self, mock_logger, mock_connect):
|
||||
mock_connect.side_effect = Exception(EXCEPTION_STR)
|
||||
self.call(
|
||||
backup.CmdBackup(),
|
||||
"",
|
||||
"You have scheduled backups to run every 86400 seconds.",
|
||||
caller=self.char1,
|
||||
)
|
||||
mock_logger.assert_called_with(f"Backup failed: {EXCEPTION_STR}")
|
||||
|
||||
@mocked_db_setting_postgres
|
||||
@patch("subprocess.run")
|
||||
@patch("evennia.utils.logger.log_sec")
|
||||
def test_postgres_success(self, mock_logger, mock_run):
|
||||
mock_run.return_value.returncode = 0
|
||||
|
||||
self.call(
|
||||
backup.CmdBackup(),
|
||||
"",
|
||||
"You have scheduled backups to run every 86400 seconds.",
|
||||
caller=self.char1,
|
||||
)
|
||||
mock_logger.assert_called_with(f"|wpostgresql db backed up in: {backup.BACKUP_FOLDER}|n")
|
||||
|
||||
@mocked_db_setting_postgres
|
||||
@patch("subprocess.run")
|
||||
@patch("evennia.utils.logger.log_err")
|
||||
def test_postgres_failure(self, mock_logger, mock_run):
|
||||
mock_run.return_value.returncode = 0
|
||||
mock_run.side_effect = Exception(EXCEPTION_STR)
|
||||
|
||||
self.call(
|
||||
backup.CmdBackup(),
|
||||
"",
|
||||
"You have scheduled backups to run every 86400 seconds.",
|
||||
caller=self.char1,
|
||||
)
|
||||
mock_logger.assert_called_with(f"Backup failed: {EXCEPTION_STR}")
|
||||
|
||||
@mocked_db_setting_sqlite
|
||||
@patch("sqlite3.connect")
|
||||
@patch("shutil.copy")
|
||||
@patch("evennia.utils.logger.log_sec")
|
||||
def test_sqlite_force_no_existing_script(self, mock_logger, mock_copy, mock_connect):
|
||||
"""backup/force should create the script if it doesn't exist, then run it."""
|
||||
mock_cursor = mock_connect.return_value.cursor.return_value
|
||||
mock_cursor.fetchone.return_value = ("ok",)
|
||||
self.call(
|
||||
backup.CmdBackup(),
|
||||
"/force",
|
||||
"You have scheduled backups to run every 86400 seconds.",
|
||||
caller=self.char1,
|
||||
)
|
||||
mock_logger.assert_called_with(f"|wsqlite3 db backed up in: {backup.BACKUP_FOLDER}|n")
|
||||
|
||||
@mocked_db_setting_unsupported
|
||||
def test_unsupported_engine(self):
|
||||
"""Unsupported DB engine should message the caller and not attempt a backup."""
|
||||
self.call(
|
||||
backup.CmdBackup(),
|
||||
"/force",
|
||||
"Database backup failed: unsupported engine 'django.db.backends.oracle'. Contrib supports postgres and sqlite3.",
|
||||
caller=self.char1,
|
||||
)
|
||||
|
||||
@mocked_db_setting_postgres
|
||||
@patch("subprocess.run")
|
||||
@patch("evennia.utils.logger.log_err")
|
||||
def test_postgres_failure_cleans_up(self, mock_logger, mock_run):
|
||||
"""A failed pg_dump should not leave a partial .sql file behind."""
|
||||
self.call(
|
||||
backup.CmdBackup(),
|
||||
"",
|
||||
"You have scheduled backups to run every 86400 seconds.",
|
||||
caller=self.char1,
|
||||
)
|
||||
mock_run.side_effect = subprocess.CalledProcessError(1, "pg_dump")
|
||||
self.call(backup.CmdBackup(), "/force", "", caller=self.char1)
|
||||
sql_files = (
|
||||
[f for f in os.listdir(backup.BACKUP_FOLDER) if f.endswith(".sql")]
|
||||
if os.path.exists(backup.BACKUP_FOLDER)
|
||||
else []
|
||||
)
|
||||
self.assertEqual(sql_files, [])
|
||||
|
||||
@mocked_db_setting_sqlite
|
||||
@patch("os.remove")
|
||||
@patch("sqlite3.connect")
|
||||
@patch("evennia.utils.logger.log_sec")
|
||||
def test_sqlite_corrupt_backup_cleans_up(self, mock_logger, mock_connect, mock_remove):
|
||||
"""A corrupt sqlite backup should be deleted and log a failure."""
|
||||
self.call(
|
||||
backup.CmdBackup(),
|
||||
"",
|
||||
"You have scheduled backups to run every 86400 seconds.",
|
||||
caller=self.char1,
|
||||
)
|
||||
mock_connect.return_value.backup.side_effect = sqlite3.DatabaseError("corrupt")
|
||||
self.call(backup.CmdBackup(), "/force", "", caller=self.char1)
|
||||
mock_remove.assert_called_once()
|
||||
mock_logger.assert_called_with(f"|rsqlite3 db backup to {backup.BACKUP_FOLDER} failed|n")
|
||||
Reference in New Issue
Block a user