forked from varia/multifeeder
62 lines
1.7 KiB
Python
62 lines
1.7 KiB
Python
|
from os import environ, mkdir
|
||
|
|
from os.path import exists
|
||
|
|
from pathlib import Path
|
||
|
|
from logging import DEBUG, INFO, basicConfig, getLogger
|
||
|
|
from json import dumps, loads
|
||
|
|
|
||
|
|
class SimpleDatabase(dict):
|
||
|
|
"""A simple database.
|
||
|
|
It is a dictionary which saves to disk on all writes. It is optimised for
|
||
|
|
ease of hacking and accessibility and not for performance or efficiency.
|
||
|
|
|
||
|
|
Written by decentral1se, as part of:
|
||
|
|
https://git.vvvvvvaria.org/decentral1se/xbotlib/src/branch/main/xbotlib.py
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self, filename, log, *args, **kwargs):
|
||
|
|
"""Initialise the object."""
|
||
|
|
self.filename = Path(filename).absolute()
|
||
|
|
self.log = getLogger(__name__)
|
||
|
|
|
||
|
|
self._loads()
|
||
|
|
self.update(*args, **kwargs)
|
||
|
|
|
||
|
|
def _loads(self):
|
||
|
|
"""Load the database."""
|
||
|
|
if not exists(self.filename):
|
||
|
|
return
|
||
|
|
|
||
|
|
try:
|
||
|
|
with open(self.filename, "r") as handle:
|
||
|
|
self.update(loads(handle.read()))
|
||
|
|
except Exception as exception:
|
||
|
|
message = f"Loading file storage failed: {exception}"
|
||
|
|
self.log.error(message, exc_info=exception)
|
||
|
|
exit(1)
|
||
|
|
|
||
|
|
def _dumps(self):
|
||
|
|
"""Save the databse to disk."""
|
||
|
|
try:
|
||
|
|
with open(self.filename, "w") as handle:
|
||
|
|
handle.write(dumps(self, indent=4, sort_keys=True))
|
||
|
|
except Exception as exception:
|
||
|
|
message = f"Saving file storage failed: {exception}"
|
||
|
|
self.log.error(message, exc_info=exception)
|
||
|
|
exit(1)
|
||
|
|
|
||
|
|
def __setitem__(self, key, val):
|
||
|
|
"""Write data to the database."""
|
||
|
|
super().__setitem__(key, val)
|
||
|
|
self._dumps()
|
||
|
|
|
||
|
|
def __delitem__(self, key):
|
||
|
|
"""Remove data from the database."""
|
||
|
|
super().__delitem__(key)
|
||
|
|
self._dumps()
|
||
|
|
|
||
|
|
def update(self, *args, **kwargs):
|
||
|
|
"""Update the database."""
|
||
|
|
for k, v in dict(*args, **kwargs).items():
|
||
|
|
self[k] = v
|
||
|
|
self._dumps()
|