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
|