signal api: add some group features

apitypes.py

@@ -43,6 +43,24 @@ class UsernameSetResponse(BaseModel):
     username: str
     username_link: str
 
+GroupLinkPolicy = Literal["disabled", "enabled", "enabled-with-approval"]
+GroupPermissionPolicy = Literal["only-admins", "every-member"]
+
+class GroupPermissions(BaseModel):
+    add_members: GroupPermissionPolicy
+    edit_group: GroupPermissionPolicy
+
+class CreateGroupRequest(BaseModel):
+    description: str = ""
+    expiration_time: int = 0 # Expiration time of messages
+    group_link: GroupLinkPolicy = "disabled"
+    members: List[str]
+    name: str
+    permissions: GroupPermissions = GroupPermissions(add_members="only-admins", edit_group="only-admins")
+
+class CreateGroupResponse(BaseModel):
+    id: str
+
 def test_reaction_message():
     data = """{
         "envelope": {

main.py

@@ -2,7 +2,7 @@
 
 import requests
 import json
-from result import Result, Ok, Err
+from result import Result, Ok, Err, is_err
 from typing import List
 from websockets.asyncio.client import connect
 import websockets
@@ -56,6 +56,73 @@ class SignalAPI:
             print("Failed to get group")
             print(r.json())
 
+    def create_group(self, group_request: CreateGroupRequest) -> Result[CreateGroupResponse, str]:
+        r = requests.post(f"{self.apiurl}/v1/groups/{self.number}", json=group_request.model_dump())
+
+        if r.status_code == 201:
+            return parse_response(CreateGroupResponse, r.text)
+        else:
+            print("Failed to create group")
+            print(r.json())
+
+    def add_group_members(self, group_id: str, numbers_to_add: List[str]) -> Result[None, str]:
+        # put request to add group members
+        # /v1/groups/{group_id}/members
+        # {
+        #    "members": ["+49123456789", "+49123456780"]
+        # }
+        r = requests.post(f"{self.apiurl}/v1/groups/{self.number}/{group_id}/members", json={"members": numbers_to_add})
+
+        if r.status_code == 204:
+            return Ok(None)
+        else:
+            print("Failed to add group members")
+            print(r.json())
+
+    def remove_group_members(self, group_id: str, numbers_to_remove: List[str]) -> Result[None, str]:
+        # delete request to remove group members
+        # /v1/groups/{group_id}/members
+        # {
+        #    "members": ["+49123456789", "+49123456780"]
+        # }
+        r = requests.delete(f"{self.apiurl}/v1/groups/{self.number}/{group_id}/members", json={"members": numbers_to_remove})
+
+        if r.status_code == 204:
+            return Ok(None)
+        else:
+            print("Failed to remove group members")
+            print(r.json())
+
+    def update_group_members(self, group_id: str, other_members: List[str]) -> Result[None, str]:
+        group = self.get_group(group_id)
+        if is_err(group):
+            return Err(group.unwrap_err())
+
+        current_members = group.unwrap().members
+
+        members_to_add = []
+        members_to_remove = []
+
+        for member in other_members:
+            if member not in current_members:
+                members_to_add.append(member)
+
+        for member in current_members:
+            if member not in other_members:
+                members_to_remove.append(member)
+
+        if len(members_to_add) > 0:
+            add_result = self.add_group_members(group_id, members_to_add)
+            if add_result.is_err():
+                return Err(add_result.unwrap_err())
+
+        if len(members_to_remove) > 0:
+            remove_result = self.remove_group_members(group_id, members_to_remove)
+            if remove_result.is_err():
+                return Err(remove_result.unwrap_err())
+
+        return Ok(None)
+
     def get_identities(self) -> Result[List[IdentityEntry], str]:
         r = requests.get(f"{self.apiurl}/v1/identities/{self.number}")