Database with Json

General Showcase
1

I have taken the source code from replit db And i have made it into a database thats connected to a JSON File like so:

import os
import unittest
import json
import requests
from replit.database import AsyncDatabase
from replit import db



backup_file_path = 'backup.json'


def create_backup():
    """Create a backup of the Replit database and sync it."""
    backup_data = dict(db)
  
    with open(backup_file_path, 'w') as file:
        json.dump(backup_data, file, indent=2)
  
    sync_backup()


def load_backup():
    """Load the backup data into the Replit database."""
    if os.path.exists(backup_file_path):
        with open(backup_file_path, 'r') as file:
            backup_data = json.load(file)
        db.update(backup_data)


def save_backup():
    """Save the Replit database data to the backup file."""
    backup_data = dict(db)
    
    with open(backup_file_path, 'w') as file:
        json.dump(backup_data, file)


def sync_backup():
    """Sync the backup file with the Replit database."""
    with open(backup_file_path, 'r') as file:
        backup_data = json.load(file)
    
    db.update(backup_data)
    

class SyncDatabase(dict):
    """Subclass of dict that syncs with a JSON backup file."""

    def __setitem__(self, key, value):
        """Set a key-value pair and sync with the backup file."""
        super().__setitem__(key, value)
        sync_backup()

    def __delitem__(self, key):
        """Delete a key-value pair and sync with the backup file."""
        super().__delitem__(key)
        sync_backup()


class TestSyncDatabase(unittest.TestCase):
    """Integration tests for syncing a backup file with the Replit database."""

    def setUp(self) -> None:
        """Grab a JWT for all the tests to share."""
        if "REPLIT_DB_URL" in os.environ:
            self.db = AsyncDatabase(os.environ["REPLIT_DB_URL"])
        else:
            password = os.environ["PASSWORD"]
            req = requests.get(
                "https://database-test-jwt.kochman.repl.co", auth=("test", password)
            )
            url = req.text
            self.db = AsyncDatabase(url)

        # nuke whatever is already here
        for k in self.db.keys():
            del self.db[k]

    def tearDown(self) -> None:
        """Nuke whatever the test added."""
        for k in self.db.keys():
            del self.db[k]

    def test_sync_backup(self) -> None:
        """Test syncing the backup file with the Replit database."""
        # Create a backup
        backup_data = dict(self.db)
        with open(backup_file_path, 'w') as file:
            json.dump(backup_data, file, indent=2)

        # Sync the backup with the Replit database
        with open(backup_file_path, 'r') as file:
            backup_data = json.load(file)
        self.db.update(backup_data)

        # Perform necessary tests
        self.assertEqual(self.db["test-key"], "value")

        # Update the Replit database and save the changes to the backup file
        self.db["test-key"] = "new-value"
        backup_data = dict(self.db)
        with open(backup_file_path, 'w') as file:
            json.dump(backup_data, file)

        # Sync the backup with the Replit database again
        with open(backup_file_path, 'r') as file:
            backup_data = json.load(file)
        self.db.update(backup_data)

        # Perform necessary tests again
        self.assertEqual(self.db["test-key"], "new-value")


if __name__ == '__main__':
    db = SyncDatabase(db)
    create_backup()
    unittest.main()

to init the backup_file, just do

create_backup()

to load the file:

load_backup()

to save:

save_backup()

to sync the data:

sync_backup()

Just add data to your database like so and save it!

create_backup()
db['Hello'] = 'hi!'
save_backup()

Pretty cool, huh?

1 Like