add management controllers

src/swibots/api/community/community_client.py

@@ -7,7 +7,7 @@
 from swibots.utils.ws.asyncstomp.async_ws_subscription import AsyncWsSubscription
 from swibots.utils.ws.common.ws_message import WsMessage
 from swibots.types import EventType
-from .controllers import ChannelController, GroupController, CommunityController
+from .controllers import ChannelController, GroupController, CommunityController, RolesController, PermissionController, RoleMemberController
 import swibots
 
 
@@ -24,6 +24,9 @@ def __init__(
         self._channels = None
         self._groups = None
         self._communities = None
+        self._roles = None
+        self._rolemember = None
+        self._permission = None
         self._ws: SwitchWSAsyncClient = None
         self._started = False
 
@@ -48,6 +51,27 @@ def communities(self) -> CommunityController:
             self._communities = CommunityController(self)
         return self._communities
 
+    @property
+    def roles(self) -> RolesController:
+        """Gets the roles controller"""
+        if self._roles is None:
+            self._roles = RolesController(self)
+        return self._roles
+
+    @property
+    def permission(self) -> PermissionController:
+        """Gets the permission controller"""
+        if self._permission is None:
+            self._permission = PermissionController(self)
+        return self._permission
+
+    @property
+    def rolemember(self) -> RoleMemberController:
+        """Gets the roles controller"""
+        if self._rolemember is None:
+            self._rolemember = RoleMemberController(self)
+        return self._rolemember
+
     @property
     def ws(self) -> SwitchWSAsyncClient:
         if self._ws is None:

src/swibots/api/community/controllers/__init__.py

@@ -1,3 +1,6 @@
 from .community_controller import *
 from .channel_controller import *
 from .group_controller import *
+from .roles_controller import *
+from .permissions_controller import *
+from .rolemember_controller import *

src/swibots/api/community/controllers/community_controller.py

@@ -14,7 +14,16 @@ class CommunityController:
     def __init__(self, client: "CommunityClient"):
         self.client = client
 
-    async def get_community(self, community_id: str):
-        """Get a community by id"""
-        response = await self.client.get(f"{BASE_PATH}?communityId={community_id}")
+    async def get_community(self, community_id: str = '', username: str = ''):
+        """Get a community by id or username"""
+        if not (community_id or username):
+            raise ValueError('community_id or username must be provided.')
+
+        if community_id and username:
+            raise ValueError("community_id and username can't be provided together.")
+        if username:
+            request_url = f"{BASE_PATH}/communityusername?communityId={community_id}"
+        else:
+            request_url = f"{BASE_PATH}?communityId={community_id}"
+        response = await self.client.get(request_url)
         return self.client.build_object(Community, response.data.get("result"))

src/swibots/api/community/controllers/permissions_controller.py

@@ -0,0 +1,72 @@
+import logging
+from typing import TYPE_CHECKING
+
+from swibots.api.community.models import Role, RolePermission
+
+
+if TYPE_CHECKING:
+    from swibots.api.community import CommunityClient
+
+log = logging.getLogger(__name__)
+
+BASE_PATH = "/v1/community/permission"
+
+
+class PermissionController:
+    def __init__(self, client: "CommunityClient"):
+        self.client = client
+
+    async def get_permission(self, role_id: str) -> RolePermission | None:
+        """Get permission by role id"""
+        response = await self.client.get(f"{BASE_PATH}/get/?roleId={role_id}")
+        return self.client.build_object(RolePermission, response.data.get("result"))
+
+    async def add_permission(self, permission: RolePermission) -> bool:
+        """Add permission"""
+        response = await self.client.post(
+            f"{BASE_PATH}/add", data=permission.to_json_request()
+        )
+        return response.data.get("success", False)
+
+    async def update_permission(
+        self,
+        permission: RolePermission = None,
+        permission_id: int = 0,
+        role_id: int = 0,
+    ) -> bool:
+        """update permission"""
+        if id:
+            permission.id = permission_id
+        if role_id:
+            permission.role_id = role_id
+
+        await self.client.put(
+            f"{BASE_PATH}/update",
+            data=permission.to_json_request(),
+        )
+        return True
+
+    async def delete_permission(self, permission_id: int) -> bool:
+        """delete permission by id"""
+        response = await self.client.delete(f"{BASE_PATH}/delete/{permission_id}")
+        return response.data.get("success", False)
+
+    async def add_user_permission(
+        self,
+        community_id: str,
+        user_id: int,
+        role_colour: str,
+        role_name: str,
+        permission: RolePermission,
+    ) -> bool:
+        resp = await self.client.post(
+            f"{BASE_PATH}/user",
+            data={
+                "communityId": community_id,
+                "memberId": user_id,
+                "roleColour": role_colour,
+                "roleName": role_name,
+                "rolePermission": permission.to_json()["rolePermission"],
+            },
+        )
+        return resp.data.get("success", False)

src/swibots/api/community/controllers/rolemember_controller.py

@@ -0,0 +1,64 @@
+import logging
+from typing import TYPE_CHECKING, Optional, List
+
+from swibots.api.common.models import User
+from swibots.api.community.models import RoleMember
+
+
+if TYPE_CHECKING:
+    from swibots.api.community import CommunityClient
+
+log = logging.getLogger(__name__)
+
+BASE_PATH = "/v1/community/member"
+
+
+class RoleMemberController:
+    def __init__(self, client: "CommunityClient"):
+        self.client = client
+
+    async def get_members(self, role_id: str) -> List[RoleMember]:
+        """Get all members by role id"""
+        response = await self.client.get(f"{BASE_PATH}/getAll?roleId={role_id}")
+        return self.client.build_list(RoleMember, response.data.get("result"))
+
+    async def add_member_to_role(
+        self,
+        community_id: str,
+        member_id: int,
+        role_id: int,
+        user_id: int,
+        user: User,
+        id: Optional[int] = None,
+    ) -> bool:
+        member = RoleMember(
+            app=self.client.app,
+            id=id,
+            member_id=member_id,
+            community_id=community_id,
+            user_id=user_id,
+            role_id=role_id,
+            user=user
+        )
+        response = await self.client.post(f"{BASE_PATH}/add", data=member.to_json_request())
+        return response.data.get("success", False)
+
+
+    async def delete_role_member(
+        self,
+        id: Optional[int] = None,
+        member_id: Optional[int] = None,
+        role_id: Optional[int] = None,
+    ) -> bool:
+        """delete member from role by id"""
+        if id:
+            response = await self.client.delete(f"{BASE_PATH}/delete/{id}")
+        elif member_id and role_id:
+            response = await self.client.delete(
+                f"{BASE_PATH}/delete?memberId={member_id}&roleId={role_id}"
+            )
+        else:
+            raise ValueError(
+                "Either provide 'id' or 'member_id with role_id' to  delete member from the role."
+            )
+        return response.data.get("success", False)

src/swibots/api/community/controllers/roles_controller.py

@@ -0,0 +1,65 @@
+import logging
+from typing import TYPE_CHECKING
+
+from swibots.api.community.models import Role, RolePermission
+
+
+if TYPE_CHECKING:
+    from swibots.api.community import CommunityClient
+
+log = logging.getLogger(__name__)
+
+BASE_PATH = "/v1/community/roles"
+
+
+class RolesController:
+    def __init__(self, client: "CommunityClient"):
+        self.client = client
+
+    async def get_roles(self, community_id: str):
+        """Get all roles by community id"""
+        response = await self.client.get(
+            f"{BASE_PATH}/getAll?communityId={community_id}"
+        )
+        return self.client.build_list(Role, response.data.get("result"))
+
+    async def add_role(
+        self, community_id: str, role_colour: str, role_name: str
+    ) -> Role:
+        """Add role to the community"""
+        role = Role(
+            app=self.client.app,
+            community_id=community_id,
+            colour=role_colour,
+            name=role_name,
+        )
+        await self.client.post(
+            f"{BASE_PATH}/add?communityId={community_id}",
+            data=role.to_json_request(),
+        )
+        return role
+
+    async def update_role(
+        self,
+        community_id: str,
+        role_id: int,
+        role_colour: str = None,
+        role_name: str = None,
+    ) -> bool:
+        """update role by community and role id"""
+        await self.client.put(
+            f"{BASE_PATH}/update",
+            data=Role(
+                app=self.client.app,
+                colour=role_colour,
+                community_id=community_id,
+                name=role_name,
+                id=role_id,
+            ).to_json_request(),
+        )
+        return True
+
+    async def delete_role(self, roleId: int):
+        """delete role by id"""
+        response = await self.client.delete(f"{BASE_PATH}/delete/{roleId}")
+        return response.data.get("success")

src/swibots/api/community/methods/__init__.py

@@ -1,11 +1,16 @@
 from .get_channel import GetChannel
 from .get_community import GetCommunity
 from .get_group import GetGroup
-
+from .permission import PermissionMethods
+from .roles import RoleMethods
+from .rolemember import RoleMemberMethods
 
 class CommunityMethods(
     GetChannel,
     GetCommunity,
     GetGroup,
+    RoleMethods,
+    PermissionMethods,
+    RoleMemberMethods
 ):
     pass

src/swibots/api/community/methods/get_community.py

@@ -4,18 +4,20 @@
 
 
 class GetCommunity:
-    async def get_community(self: "swibots.ApiClient", id: str) -> Community:
-        """Get a community by its ID
+    async def get_community(self: "swibots.ApiClient", id: str = '', username: str = '') -> Community:
+        """Get a community by its ID or username
 
         Args:
             id (`str`): The ID of the community
+            username (`str`): Username of the community
 
         Returns:
             :obj:`swibots.api.community.models.Community`: The community object.
 
         Raises:
             :obj:`switch.error.SwitchError`: If the community could not be retrieved
+            :obj:`ValueError`: if the community id and username is provided together.
 
         This method does the same as :meth:`switch.api.community.controllers.CommunityController.get_community`.
         """
-        return await self.community_service.communities.get_community(id)
+        return await self.community_service.communities.get_community(id, username=username)