PrivateCoffee/matrix-roomba
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(bot): combine user lock/unlock into single method

Browse source 
Refactored `lock_user` method to support both locking and unlocking users by introducing an optional `locked` parameter. This simplifies the codebase and removes redundancy by merging the previously separate `lock_user` and `unlock_user` methods. Improves maintainability and consistency in user management operations.
This commit is contained in:
Kumi 2024-09-01 09:21:19 +02:00
parent b2cc13b72d
commit bcf63c3a11
Signed by: kumi
GPG key ID: ECBCC9082395383F
1 changed files with 8 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

42
src/matrix_roomba/bot.py
View file

@ -224,57 +224,31 @@ class RoombaBot:
f"Failed to shutdown room {room_id}.",
)
async def lock_user(self, user_id):
"""Lock a user.
async def lock_user(self, user_id, locked=True):
"""Lock or unlock a user.
Args:
user_id (str): The user ID to lock.
locked (bool, optional): Whether to lock (True) or unlock (False) the user. Defaults to True.
"""
url = f"{self.homeserver}/_synapse/admin/v2/users/{user_id}"
headers = {
"Authorization": f"Bearer {self.access_token}",
"Content-Type": "application/json",
}
payload = {"locked": True}
payload = {"locked": locked}
async with aiohttp.ClientSession() as session:
async with session.put(url, headers=headers, json=payload) as resp:
if resp.status == 200:
self.logger.debug(f"User {user_id} locked successfully")
self.logger.debug(f"User {user_id} {"locked" if locked else "unlocked"} successfully")
await self.send_message(
self.moderation_room_id, f"User {user_id} locked successfully."
self.moderation_room_id, f"User {user_id} {"locked" if locked else "unlocked"} successfully."
)
else:
self.logger.error(f"Failed to lock user {user_id}: {resp.status}")
self.logger.error(f"Failed to {"lock" if locked else "unlock"} user {user_id}: {resp.status}")
await self.send_message(
self.moderation_room_id, f"Failed to lock user {user_id}."
)
async def unlock_user(self, user_id):
"""Unlock a user.
Args:
user_id (str): The user ID to unlock.
"""
url = f"{self.homeserver}/_synapse/admin/v2/users/{user_id}"
headers = {
"Authorization": f"Bearer {self.access_token}",
"Content-Type": "application/json",
}
payload = {"locked": False}
async with aiohttp.ClientSession() as session:
async with session.put(url, headers=headers, json=payload) as resp:
if resp.status == 200:
self.logger.debug(f"User {user_id} unlocked successfully")
await self.send_message(
self.moderation_room_id,
f"User {user_id} unlocked successfully.",
)
else:
self.logger.error(f"Failed to unlock user {user_id}: {resp.status}")
await self.send_message(
self.moderation_room_id, f"Failed to unlock user {user_id}."
self.moderation_room_id, f"Failed to {"lock" if locked else "unlock"} user {user_id}."
)
async def send_message(self, room_id, message):