Add backup up of data.

Author: benwoo1110

.gitignore

@@ -1,4 +1,5 @@
 # Custom
+backup/*
 data.json
 
 # Byte-compiled / optimized / DLL files

bot.py

@@ -1,6 +1,7 @@
 import logging
 from pathlib import Path
 
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from dis_snek.client import Snake
 from dis_snek.errors import CommandCheckFailure
 from dis_snek.mixins.send import SendMixin
@@ -76,6 +77,11 @@ async def on_command_error(self, source: str, error: Exception, *args, **kwargs)
 
 
 bot = BytehackzBot(default_prefix="!", debug_scope=GUILD)
-bot.storage = JsonStorage(bot, "data.json")
+
+bot.storage = JsonStorage("data.json", "./backup", 20)
+scheduler = AsyncIOScheduler()
+scheduler.add_job(bot.storage.backup, 'interval', minutes=5)
+scheduler.start()
+
 bot.load_all_scales("modules")
 bot.start(TOKEN)

requirements.txt

@@ -2,3 +2,4 @@ python-dotenv==0.19.1
 orjson==3.6.4
 numpy==1.21.4
 dis-snek==2.0.0
+apscheduler==3.8.1

storage/storage.py

@@ -1,3 +1,5 @@
+import datetime
+import os
 from pathlib import Path
 
 import attr
@@ -15,9 +17,10 @@ class Container(DictSerializationMixin):
 
 
 class JsonStorage:
-    def __init__(self, bot, filename: str):
-        self.bot = bot
+    def __init__(self, filename: str, backup_folder: str, max_backups=5):
         self.filename = Path(filename)
+        self.backup_folder = Path(backup_folder)
+        self.max_backups = max_backups
         self.container = None
         self._init_data()
 
@@ -29,7 +32,23 @@ def _init_data(self):
         else:
             self.container = Container()
 
+        self.backup_folder.mkdir(exist_ok=True)
+
     def save(self):
-        with open(self.filename, "wb") as file:
+        self._save_file(self.filename)
+
+    def backup(self):
+        backup_filename = f"backup-{datetime.datetime.now().timestamp()}.json"
+        backup_path = self.backup_folder.joinpath(backup_filename)
+        self._save_file(backup_path)
+
+        backup_files = sorted(os.listdir(self.backup_folder), key=lambda file: os.path.getctime(self.backup_folder.joinpath(file).absolute()))
+        if len(backup_files) > self.max_backups:
+            os.remove(self.backup_folder.joinpath(backup_files[0]).absolute())
+
+        print("Backup done")
+
+    def _save_file(self, path):
+        with open(path, "wb") as file:
             data = orjson.dumps(self.container.to_dict())
             file.write(data)