Utilities and JSON Database support #51
50
discord/ext/json/database.py
Normal file
50
discord/ext/json/database.py
Normal file
@@ -0,0 +1,50 @@
|
||||
from .types import JSONFile, Entry
|
||||
from typing import Union, Any
|
||||
|
||||
class Database:
|
||||
def __init__(self, file:JSONFile):
|
||||
self.file = file;
|
||||
self.database = {};
|
||||
|
||||
async def getData(self) -> dict:
|
||||
"""
|
||||
Get contents from a JSONFile
|
||||
"""
|
||||
|
||||
contents = await self.file.serialize("load")
|
||||
|
||||
return contents
|
||||
|
||||
async def dumpData(self, data:dict) -> None:
|
||||
"""
|
||||
Dump a dict into file
|
||||
=====
|
||||
|
||||
data `dict` -
|
||||
|
||||
The data to be dumped into the file.
|
||||
"""
|
||||
|
||||
await self.file.serialize("dump", contents=data)
|
||||
|
||||
async def loadFile(self) -> None:
|
||||
"""
|
||||
Load JSON from file to self.database
|
||||
"""
|
||||
|
||||
self.database = await self.getData();
|
||||
|
||||
async def getEntry(self, name:str) -> Entry:
|
||||
value : Union[Any] = self.database[name]
|
||||
|
||||
return Entry(name, value)
|
||||
|
||||
async def editEntry(self, name:str, value:Union[Any]):
|
||||
self.database[name] = value;
|
||||
|
||||
async def saveData(self):
|
||||
"""
|
||||
Save current database to file
|
||||
"""
|
||||
|
||||
await self.file.serialize("dump", contents=self.database)
|
||||
@@ -2,17 +2,6 @@ import json
|
||||
|
||||
from typing import Union, Any
|
||||
|
||||
class Serializer:
|
||||
def bind_file(self, file:str):
|
||||
"""Bind file to be serialized(loaded or dumped)"""
|
||||
|
||||
self.file = file
|
||||
|
||||
async def load(self, content:str):
|
||||
raise NotImplementedError("Serializer has to be implemented!")
|
||||
|
||||
async def dump(self):
|
||||
raise NotImplementedError("Serializer hasn't been implemented")
|
||||
|
||||
class File:
|
||||
def __init__(self, path:str):
|
||||
@@ -92,4 +81,57 @@ class File:
|
||||
|
||||
return contents
|
||||
|
||||
async def serialize(self, serializer:Serializer) -> Union[Any]:
|
||||
async def serialize(self, serializer, *args, **kwargs) -> Union[Any]:#
|
||||
return serializer(*args, **kwargs)
|
||||
|
||||
class JSONFile(File):
|
||||
load = json.load
|
||||
dump = json.dump
|
||||
|
||||
async def serialize(self, serializer:str, *args, **kwargs) -> Union[dict, int]:
|
||||
"""
|
||||
Serialize JSON data
|
||||
=====
|
||||
|
||||
serializer `str`:
|
||||
|
||||
dump or load data, set to "load" for loading file, "dump" to dump json to file
|
||||
|
||||
contents `dict` of kwargs:
|
||||
|
||||
what to dump.
|
||||
"""
|
||||
|
||||
if not serializer:
|
||||
raise ValueError("Argument serializer should be either load or dump, got NoneType instead.")
|
||||
|
||||
if serializer == "load":
|
||||
reader = self.open_reader()
|
||||
contents = JSONFile.load(reader)
|
||||
reader.close()
|
||||
|
||||
return contents
|
||||
elif serializer == "dump":
|
||||
writer = self.open_writer()
|
||||
|
||||
JSONFile.dump(kwargs["contents"], writer)
|
||||
|
||||
writer.close()
|
||||
|
||||
return 0;
|
||||
|
||||
class Entry:
|
||||
"""
|
||||
Class representing a JSON Entry
|
||||
"""
|
||||
|
||||
def __init__(self, name:str, value:Union[Any]):
|
||||
self.name : str = name;
|
||||
self.value : Union[Any] = value;
|
||||
|
||||
async def browse(self, name:str):
|
||||
"""
|
||||
Browse entry and get a new Entry object if it finds something successfully
|
||||
"""
|
||||
|
||||
return Entry(name, self.value[name])
|
||||
Reference in New Issue
Block a user