Multifeeding RSS streams into points of access.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

62 lines
1.9 KiB

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()