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.
92 lines
2.3 KiB
92 lines
2.3 KiB
from typing import Any
|
|
from typing import Dict
|
|
from typing import List
|
|
from typing import Tuple
|
|
from typing import Union
|
|
|
|
try:
|
|
from prettytable import PrettyTable
|
|
PRETTYTABLE = True
|
|
except ImportError:
|
|
PRETTYTABLE = False
|
|
|
|
from pysondb.db_types import DBSchemaType
|
|
from pysondb.db_types import SingleDataType
|
|
|
|
|
|
OldDataType = Dict[str, List[Dict[str, Any]]]
|
|
NewDataType = Dict[
|
|
str,
|
|
Union[
|
|
int,
|
|
str,
|
|
List[str],
|
|
Dict[
|
|
str,
|
|
Any
|
|
]
|
|
]
|
|
]
|
|
|
|
|
|
def migrate(old_db_data: OldDataType) -> NewDataType:
|
|
new_data: NewDataType = {'version': 2, 'keys': [], 'data': {}}
|
|
|
|
if not old_db_data['data']:
|
|
return new_data
|
|
|
|
new_data['keys'] = list(old_db_data['data'][0].keys())
|
|
if isinstance(new_data['keys'], list):
|
|
new_data['keys'].remove('id')
|
|
if isinstance(new_data['data'], dict):
|
|
for d in old_db_data['data']:
|
|
_id = str(d['id'])
|
|
del d['id']
|
|
new_data['data'][_id] = d
|
|
|
|
return new_data
|
|
|
|
|
|
def print_db_as_table(data: NewDataType) -> Tuple[str, int]:
|
|
if not PRETTYTABLE:
|
|
return 'install prettytable (pip3 install prettytable) to run the following command', 1
|
|
if 'version' not in data:
|
|
return 'the DB must be a v2 DB, you can use the migrate command to the convert your DB', 1
|
|
|
|
if isinstance(data['keys'], list):
|
|
keys = sorted(data['keys'])
|
|
x = PrettyTable()
|
|
x.field_names = ['id', *keys]
|
|
if isinstance(data['data'], dict):
|
|
for id, values in data['data'].items():
|
|
x.add_row([id, *[values[i] for i in keys]])
|
|
|
|
return x.get_string(), 0
|
|
|
|
return '', 0
|
|
|
|
|
|
def merge_n_db(*dbs: DBSchemaType) -> Tuple[DBSchemaType, str, int]:
|
|
keys: List[str] = []
|
|
new_db: DBSchemaType = {}
|
|
data: Dict[str, SingleDataType] = {}
|
|
for db in dbs:
|
|
if isinstance(db['keys'], list):
|
|
if not keys:
|
|
keys = db['keys']
|
|
if db['keys'] != keys:
|
|
return {}, 'All the DB\'s must have the same keys', 1
|
|
|
|
if isinstance(db['data'], dict):
|
|
data = {**data, **db['data']}
|
|
|
|
new_db['version'] = 2
|
|
new_db['keys'] = keys
|
|
new_db['data'] = data
|
|
return new_db, '', 0
|
|
|
|
|
|
def purge_db(_: Any) -> DBSchemaType:
|
|
data: DBSchemaType = {'version': 2, 'keys': [], 'data': {}}
|
|
return data
|