Bases: object
This is account manager of the Laboratory members and chat groups.
Only registered users have access to the Lab assistant.
The user account information is stored as a dictionary in JSON format:
{user_id: password}
The chat groups information is stored as a dictionary in JSON format:
{chat_group_id: [user_id, ]}
Source code in labridge\accounts\users.py
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 | class AccountManager(object):
r"""
This is account manager of the Laboratory members and chat groups.
Only registered users have access to the Lab assistant.
The user account information is stored as a dictionary in JSON format:
`{user_id: password}`
The chat groups information is stored as a dictionary in JSON format:
`{chat_group_id: [user_id, ]}`
"""
def __init__(self):
root = Path(__file__)
for idx in range(3):
root = root.parent
self.root = root
self.user_ids_path = str(root / USER_IDS_PERSIS_PATH)
self.chat_group_ids_path = str(root / CHAT_GROUP_IDS_PERSIST_PATH)
self.fs = fsspec.filesystem("file")
dir_path = str(Path(self.user_ids_path).parent)
if not self.fs.exists(dir_path):
self.fs.makedirs(dir_path)
def _get_user_ids_dict(self) -> Dict[str, str]:
r""" Get the user account dict. """
if not self.fs.exists(self.user_ids_path):
return {}
with self.fs.open(self.user_ids_path, "rb") as f:
user_ids = json.load(f)
return user_ids
def _get_chat_group_ids_dict(self) -> Dict[str, List[str]]:
r""" Get the chat group id dict. """
if not self.fs.exists(self.chat_group_ids_path):
return {}
with self.fs.open(self.chat_group_ids_path, "rb") as f:
chat_group_ids = json.load(f)
return chat_group_ids
def get_users(self) -> List[str]:
r""" Get the registered users. """
return list(self._get_user_ids_dict().keys())
def get_chat_groups(self) -> List[str]:
r""" Get the registered chat groups """
return list(self._get_chat_group_ids_dict().keys())
def user_log_in(self, user_id: str, password: str) -> bool:
r""" User log in """
try:
self.check_valid_user(user_id)
user_ids = self._get_user_ids_dict()
return password == user_ids[user_id]
except ValueError:
return False
def check_valid_user(self, user_id: str):
r""" Check whether the given user is registered. """
user_list = self.get_users()
if user_id not in user_list:
raise ValueError(f"Invalid user id, the user {user_id} is not registered.")
def is_valid_chat_group(self, chat_group_id: str):
r""" Check whether the given chat group is registered """
chat_group_list = self.get_chat_groups()
if chat_group_id not in chat_group_list:
raise ValueError(f"The chat group {chat_group_id} is not registered.")
def add_user(self, user_id: str, password: str):
r""" Register a new user. """
user_ids = self._get_user_ids_dict()
if user_id not in user_ids:
user_ids[user_id] = password
with self.fs.open(self.user_ids_path, "w") as f:
f.write(json.dumps(user_ids))
def add_chat_group(self, chat_group_id: str, user_list: List[str]) -> Optional[str]:
r"""
Register a new chat group along with its members.
All members in the chat group should have registered as a user.
"""
for user_id in user_list:
try:
self.check_valid_user(user_id)
except ValueError as e:
return f"Error: {e!s}"
chat_group_ids = self._get_chat_group_ids_dict()
if chat_group_id not in chat_group_ids:
chat_group_ids[chat_group_id] = user_list
with self.fs.open(self.chat_group_ids_path, "w") as f:
f.write(json.dumps(chat_group_ids))
return None
def update_chat_group_members(self, chat_group_id: str, new_user_list: List[str]) -> Optional[str]:
r""" Update the members of a chat group. """
for user_id in new_user_list:
try:
self.check_valid_user(user_id)
except ValueError as e:
return f"Error: {e!s}"
chat_group_ids = self._get_chat_group_ids_dict()
chat_group_ids[chat_group_id] = new_user_list
with self.fs.open(self.chat_group_ids_path, "w") as f:
f.write(json.dumps(chat_group_ids))
return None
|
labridge.accounts.users.AccountManager.add_chat_group(chat_group_id, user_list)
Register a new chat group along with its members.
All members in the chat group should have registered as a user.
Source code in labridge\accounts\users.py
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106 | def add_chat_group(self, chat_group_id: str, user_list: List[str]) -> Optional[str]:
r"""
Register a new chat group along with its members.
All members in the chat group should have registered as a user.
"""
for user_id in user_list:
try:
self.check_valid_user(user_id)
except ValueError as e:
return f"Error: {e!s}"
chat_group_ids = self._get_chat_group_ids_dict()
if chat_group_id not in chat_group_ids:
chat_group_ids[chat_group_id] = user_list
with self.fs.open(self.chat_group_ids_path, "w") as f:
f.write(json.dumps(chat_group_ids))
return None
|
labridge.accounts.users.AccountManager.add_user(user_id, password)
Register a new user.
Source code in labridge\accounts\users.py
| def add_user(self, user_id: str, password: str):
r""" Register a new user. """
user_ids = self._get_user_ids_dict()
if user_id not in user_ids:
user_ids[user_id] = password
with self.fs.open(self.user_ids_path, "w") as f:
f.write(json.dumps(user_ids))
|
labridge.accounts.users.AccountManager.check_valid_user(user_id)
Check whether the given user is registered.
Source code in labridge\accounts\users.py
| def check_valid_user(self, user_id: str):
r""" Check whether the given user is registered. """
user_list = self.get_users()
if user_id not in user_list:
raise ValueError(f"Invalid user id, the user {user_id} is not registered.")
|
labridge.accounts.users.AccountManager.get_chat_groups()
Get the registered chat groups
Source code in labridge\accounts\users.py
| def get_chat_groups(self) -> List[str]:
r""" Get the registered chat groups """
return list(self._get_chat_group_ids_dict().keys())
|
labridge.accounts.users.AccountManager.get_users()
Get the registered users.
Source code in labridge\accounts\users.py
| def get_users(self) -> List[str]:
r""" Get the registered users. """
return list(self._get_user_ids_dict().keys())
|
labridge.accounts.users.AccountManager.is_valid_chat_group(chat_group_id)
Check whether the given chat group is registered
Source code in labridge\accounts\users.py
| def is_valid_chat_group(self, chat_group_id: str):
r""" Check whether the given chat group is registered """
chat_group_list = self.get_chat_groups()
if chat_group_id not in chat_group_list:
raise ValueError(f"The chat group {chat_group_id} is not registered.")
|
labridge.accounts.users.AccountManager.update_chat_group_members(chat_group_id, new_user_list)
Update the members of a chat group.
Source code in labridge\accounts\users.py
108
109
110
111
112
113
114
115
116
117
118
119
120 | def update_chat_group_members(self, chat_group_id: str, new_user_list: List[str]) -> Optional[str]:
r""" Update the members of a chat group. """
for user_id in new_user_list:
try:
self.check_valid_user(user_id)
except ValueError as e:
return f"Error: {e!s}"
chat_group_ids = self._get_chat_group_ids_dict()
chat_group_ids[chat_group_id] = new_user_list
with self.fs.open(self.chat_group_ids_path, "w") as f:
f.write(json.dumps(chat_group_ids))
return None
|
labridge.accounts.users.AccountManager.user_log_in(user_id, password)
User log in
Source code in labridge\accounts\users.py
| def user_log_in(self, user_id: str, password: str) -> bool:
r""" User log in """
try:
self.check_valid_user(user_id)
user_ids = self._get_user_ids_dict()
return password == user_ids[user_id]
except ValueError:
return False
|