-
Notifications
You must be signed in to change notification settings - Fork 0
/
dataCentre.py
88 lines (79 loc) · 3.07 KB
/
dataCentre.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
import pandas as pd
from magicPlanAPI import MagicPlanAPI
class DataCentre:
def __init__(self):
"""
Instance to simulate a Cache by storing values in files and reading at the start of the program into variables
"""
self.paths = {
'plans': 'resources/data/plans.json',
'credentials': 'resources/data/credentials.json',
'users': 'resources/data/users.json'
}
self.data = dict()
self.load_data()
self.connect_to_api()
def get_search_plans(self, search):
"""
Returns all plans which match the search term
If no search term is provided all the plans get returned
:param search: search term
:return: Dict of plans with dict['ID','name']
"""
if search is not None:
# searching plans for search term with case sensitivity
mask = self.data['plans']['name'].str.contains(search)
filtered_plans = self.data['plans'][self.data['plans']['name'].str.contains(search)]
plans = dict()
if len(filtered_plans) == 0:
return dict()
for index, plan in filtered_plans.iterrows():
plans[plan['id']] = plan['name']
return plans
else:
return self.get_plans()
def load_data(self):
"""
Loads all the files into class variables
"""
for datasource in self.paths:
try:
self.data[datasource] = pd.read_json(self.paths[datasource])
except FileNotFoundError:
print("Die Datei konnte nicht geöffnet werden.")
def get_plans(self):
"""
Returns all plans
:return: Dict of plans with dict['ID','name']
"""
plans = dict()
for index, plan in self.data['plans'].iterrows():
plans[plan['id']] = plan['name']
return plans
def reload_plans(self):
"""
Reloads plans by pulling them from the magicplan API
:return: Dict of plans with dict['ID','name']
"""
self.data['plans'] = pd.DataFrame(self.magic_api.get_projects(as_json=True))
return self.get_plans()
def save_data(self):
"""
Stores the data in class variables into files to save them
"""
for datasource in self.paths:
with open(self.paths[datasource], 'w') as f:
f.write(self.data[datasource].to_json(indent=2))
print("{}\t -> \t{}".format(datasource, self.paths[datasource]))
def connect_to_api(self):
"""
Initiates magicplanAPI instance with credentials if defined
:return: MagicPlanAPI instance
"""
try:
customer_id = self.data['credenitals']['customerID'][0]
private_key = self.data['credenitals']['private_key'][0]
user_email = self.data['credenitals']['user_email'][0]
self.magic_api = MagicPlanAPI(customerID=customer_id, private_key=private_key, user_email=user_email)
except:
self.magic_api = MagicPlanAPI()