-
Notifications
You must be signed in to change notification settings - Fork 0
/
handlers.py
161 lines (131 loc) · 4.31 KB
/
handlers.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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
from config import ROOT_USER
from model.database import Key
from model.exception import CannotDeleteRootUserError, InvalidCredentialsError, InvalidNumberOfParamsError, InvalidTTLValueError, NoDbSelectedError, UserNotLoggedInError, UserUnauthorizedError
from model.router import Context
def login(ctx: Context) -> None:
try:
username = ctx.params[0]
password = ctx.params[1]
except IndexError:
raise InvalidNumberOfParamsError
if not ctx.store.authenticate_user(username, password):
raise InvalidCredentialsError
ctx.username = username
ctx.database_name = str()
ctx.database = None
ctx.response = 'login: ok'
def whoami(ctx: Context) -> None:
if ctx.username == '':
raise UserNotLoggedInError
ctx.response = ctx.username
def create_db(ctx: Context) -> None:
try:
new_db_name = ctx.params[0]
except IndexError:
raise InvalidNumberOfParamsError
ctx.store.create_database(username=ctx.username, new_db_name=new_db_name)
ctx.store.add_user_to_owners(username=ROOT_USER, db_name=new_db_name)
ctx.response = 'create_db: ok'
def add_user_to_owners(ctx: Context) -> None:
try:
new_owner_username = ctx.params[0]
db_name = ctx.params[1]
except IndexError:
raise InvalidNumberOfParamsError
if ctx.username != ROOT_USER:
raise UserUnauthorizedError
ctx.store.add_user_to_owners(username=new_owner_username, db_name=db_name)
ctx.response = 'add_user_to_owners: ok'
def select_db(ctx: Context) -> None:
try:
db_name = ctx.params[0]
except IndexError:
raise InvalidNumberOfParamsError
ctx.database = ctx.store.get_database_by_name(username=ctx.username, db_name=db_name)
ctx.database_name = db_name
ctx.response = 'select_db: ok'
def current_db(ctx: Context) -> None:
if ctx.database_name == '':
raise NoDbSelectedError
ctx.response = ctx.database_name
def get(ctx: Context) -> None:
try:
key = ctx.params[0]
except IndexError:
raise InvalidNumberOfParamsError
ctx.response = ctx.database.get(key=key)
def put(ctx: Context) -> None:
try:
key = ctx.params[0]
value = ctx.params[1]
except IndexError:
raise InvalidNumberOfParamsError
ctx.database.put(key=key, value=value)
try:
ttl_string = ctx.params[2]
except IndexError:
ctx.database.remove_ttl(key=key)
else:
if not ttl_string.isdigit():
raise InvalidTTLValueError
ttl = int(ttl_string)
ctx.database.set_ttl(key=key, ttl=ttl)
ctx.response = 'put: ok'
def delete(ctx: Context) -> None:
try:
key: Key = ctx.params[0]
except IndexError:
raise InvalidNumberOfParamsError
ctx.database.delete(key=key)
ctx.response = 'delete: ok'
def update(ctx: Context) -> None:
try:
key = ctx.params[0]
value = ctx.params[1]
except IndexError:
raise InvalidNumberOfParamsError
ctx.database.update(key=key, value=value)
try:
ttl_string = ctx.params[2]
except IndexError:
ctx.database.remove_ttl(key=key)
else:
if not ttl_string.isdigit():
raise InvalidTTLValueError
ttl = int(ttl_string)
ctx.database.set_ttl(key=key, ttl=ttl)
ctx.response = 'update: ok'
def delete_user(ctx: Context) -> None:
if ctx.username != ROOT_USER:
raise UserUnauthorizedError
try:
user_to_delete = ctx.params[0]
except IndexError:
raise InvalidNumberOfParamsError
if user_to_delete == ROOT_USER:
raise CannotDeleteRootUserError
ctx.store.delete_user(user_to_delete=user_to_delete)
ctx.response = 'delete_user: ok'
def delete_db(ctx: Context) -> None:
try:
db_to_delete = ctx.params[0]
except IndexError:
raise InvalidNumberOfParamsError
ctx.store.delete_database(username=ctx.username, db_to_delete=db_to_delete)
current_selected_db_name = ctx.database_name
if current_selected_db_name == db_to_delete:
ctx.database_name = str()
ctx.database = None
ctx.response = 'delete_db: ok'
def list_users(ctx: Context) -> None:
ctx.response = str(ctx.store.list_users_of_db(db_name=ctx.database_name))
def list_dbs(ctx: Context) -> None:
ctx.response = str(ctx.store.list_dbs_of_user(username=ctx.username))
def register_user(ctx: Context) -> None:
try:
username_to_create = ctx.params[0]
password = ctx.params[1]
except IndexError:
raise InvalidNumberOfParamsError
ctx.store.create_user(username_to_create=username_to_create, password=password)
ctx.response = 'create_user: ok'