-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathdatabase.py
137 lines (120 loc) · 5.72 KB
/
database.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
from typing import List, Union
class Database(List[dict]):
"""Class that works as a list of dictionaries with attributes for each key
Database class which inherits the properties of a list of dictionaries. It
also has two extra methods and an attribute per key of the internal
dictionaries. Each key attribute is a list of the values of those keys. The
add_entry method is a wrapper for the append method that also appends the
key values to the attributes. The search method returns the Database with
only the dictionaries whose key values match the requested key values.
"""
def __init__(self, *args: dict, index: str = None):
"""Initializes the Database class to be a list of dictionaries
Initializes the Database class by calling the list __init__, setting
the name attribute to the class name, and looping through the given
dictionaries to append them to the database. Contains an optionally
indicated 'index' input which will be stored as the only required key
for future input data.
:param args: dictionaries to add to the database
:type args: dict
"""
super().__init__()
self.Name = self.__class__.__name__
self.Index = index
self.__dict__[index] = []
for arg in args:
self.add_entry(arg)
def add_entry(self, entry: dict, **kwargs) -> dict:
"""Wrapper for the list append method that adds contents to attributes
Wrapper for the append method with the added effect of appending to the
attributes and returning the added dictionary for validation. Key word
arguments may be added to add a key to the dictionary before appending
it. This will be both appended and returned. Any data added that
contains an index key that is already in the database will overwrite
the previous database item. If the item to be overwritten is a list,
the default behavior is to append to that list instead of deleting it.
:param entry: Dictionary to be appended to the database
:type entry: dict
:param kwargs: Optional key word arguments to append to each dictionary
:type kwargs: dict
:return: The appended dictionary
:rtype: dict
"""
# add extra args to the dict
for key, value in kwargs.items():
entry[key] = value
# check if new entry index matches any existing entries and overwrite
# as necessary, appending any list items within the dicts
if self.Index is not None:
if self.Index not in entry.keys():
raise KeyError("{} index key {} not found in new entry".format(
type(self).__name__, self.Index))
try:
same_id = self.search(**{self.Index: entry[self.Index]})
replace_index = self.index(same_id)
for key, value in same_id.items():
# check for appendable items/update existing info
if key not in entry.keys():
entry[key] = value
elif isinstance(value, list):
entry[key] = value + entry[key]
self[replace_index] = entry
except IndexError:
self.append(entry)
else:
self.append(entry)
# make attributes TODO: fix attributes overwriting parent methods
for key, value in entry.items():
# attributes must be remade each time since elements are mutable
vars(self)[key] = []
for item in self:
if key in item:
vars(self)[key].append(item[key])
else:
vars(self)[key].append(None)
vars(self)[key] = tuple(vars(self)[key])
return entry
def search(self, get: str = "latest", **kwargs) -> Union[dict, List[dict]]:
"""Method to return a subset of the database based on a key word value
Search method for the database class. Takes a search 'mode' in the get
parameter which can be either 'first', 'all', or 'latest' (default).
It also takes at least one more key word argument to use as a search
term. The key is the matching key in the database and the value matches
the value of that key.
:param get: Determines the search mode (first, all, latest)
:type get: str
:param kwargs: Key value pairs to search the database for
:type kwargs: dict
:return: Database subset that matches the search terms
:rtype: Database
"""
assert get in ["latest", "all", "first"]
getvals = []
if get == "latest":
the_list = self.__reversed__()
else:
the_list = self
for item in the_list:
for key, value in kwargs.items():
if key not in item.keys():
continue
elif item[key] == value:
if get in ["latest", "first"]:
return item.copy()
elif get == "all":
getvals.append(item)
if get == "all" and getvals:
return self.__class__(*getvals).copy()
else:
raise IndexError(
"No {} with the value {} found in {} database".format(
" or ".join([str(key) for key in kwargs.keys()]),
" or ".join([str(value) for value in kwargs.values()]),
type(self).__name__))
if __name__ == "__main__":
mydb = Database({"a": 1, "b": 2}, {"a": 2, "b": 3, "c": [4]},
index="a")
mydb.add_entry({"a": 1}, **{"c": [4]})
mydb.add_entry({"a": 1, "b": 3})
print(mydb.b)
print(mydb)