pangea-sdk: add AuthN groups

Author: kenany

.vscode/extensions.json

@@ -10,10 +10,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - AuthZ: `expires_at` to tuples.
+- AuthN: groups.
 
 ## 6.0.0 - 2025-04-21
 
-### Added 
+### Added
 
 - Redact: `fpe_context` on `StructuredResult`.
 - AI Guard: detector overrides.
@@ -84,7 +85,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `Audit.fix_consistency_proofs` is now a private method.
 - `pangea.deep_verify` error message to `warning` when `not_persisted` event.
 
-### Fixed 
+### Fixed
 
 - `pangea.audit_dump` only dump before events if the leaf_index is not None.
 
@@ -131,7 +132,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
-- `attributes` field in `/check` endpoint. Now it's a `Dict[str, Any]` 
+- `attributes` field in `/check` endpoint. Now it's a `Dict[str, Any]`
 
 ### Fixed
 
@@ -238,7 +239,7 @@ Note that Sanitize and Secure Share did not make it into this release.
 
 ## [3.7.0] - 2024-02-26
 
-### Added 
+### Added
 
 - Vault service. Post quantum signing algorithms support
 

.vscode/settings.json

@@ -2,14 +2,17 @@
 # Author: Pangea Cyber Corporation
 from __future__ import annotations
 
-from typing import Dict, List, Optional, Union
+from typing import Dict, List, Literal, Optional, Union
 
 import pangea.services.authn.models as m
 from pangea.asyncio.services.base import ServiceBaseAsync
 from pangea.config import PangeaConfig
 from pangea.response import PangeaResponse, PangeaResponseResult
 
-SERVICE_NAME = "authn"
+__all__ = ["AuthNAsync"]
+
+
+_SERVICE_NAME = "authn"
 
 
 class AuthNAsync(ServiceBaseAsync):
@@ -35,7 +38,7 @@ class AuthNAsync(ServiceBaseAsync):
         authn = AuthNAsync(token=PANGEA_TOKEN, config=authn_config)
     """
 
-    service_name = SERVICE_NAME
+    service_name = _SERVICE_NAME
 
     def __init__(
         self,
@@ -65,7 +68,7 @@ def __init__(
         self.agreements = AuthNAsync.AgreementsAsync(token, config, logger_name=logger_name)
 
     class SessionAsync(ServiceBaseAsync):
-        service_name = SERVICE_NAME
+        service_name = _SERVICE_NAME
 
         def __init__(
             self,
@@ -162,7 +165,7 @@ async def logout(self, user_id: str) -> PangeaResponse[m.SessionLogoutResult]:
             )
 
     class ClientAsync(ServiceBaseAsync):
-        service_name = SERVICE_NAME
+        service_name = _SERVICE_NAME
 
         def __init__(
             self,
@@ -222,7 +225,7 @@ async def jwks(
             return await self.request.post("v2/client/jwks", m.ClientJWKSResult, {})
 
         class SessionAsync(ServiceBaseAsync):
-            service_name = SERVICE_NAME
+            service_name = _SERVICE_NAME
 
             def __init__(
                 self,
@@ -359,7 +362,7 @@ async def refresh(
                 )
 
         class PasswordAsync(ServiceBaseAsync):
-            service_name = SERVICE_NAME
+            service_name = _SERVICE_NAME
 
             def __init__(
                 self,
@@ -419,7 +422,7 @@ async def expire(self, user_id: str) -> PangeaResponse[PangeaResponseResult]:
                 return await self.request.post("v2/user/password/expire", PangeaResponseResult, {"id": user_id})
 
         class TokenAsync(ServiceBaseAsync):
-            service_name = SERVICE_NAME
+            service_name = _SERVICE_NAME
 
             def __init__(
                 self,
@@ -456,7 +459,7 @@ async def check(self, token: str) -> PangeaResponse[m.ClientTokenCheckResult]:
                 )
 
     class UserAsync(ServiceBaseAsync):
-        service_name = SERVICE_NAME
+        service_name = _SERVICE_NAME
 
         def __init__(
             self,
@@ -667,7 +670,7 @@ async def list(
             return await self.request.post("v2/user/list", m.UserListResult, data=input.model_dump(exclude_none=True))
 
         class InvitesAsync(ServiceBaseAsync):
-            service_name = SERVICE_NAME
+            service_name = _SERVICE_NAME
 
             def __init__(
                 self,
@@ -739,7 +742,7 @@ async def delete(self, id: str) -> PangeaResponse[m.UserInviteDeleteResult]:
                 )
 
         class AuthenticatorsAsync(ServiceBaseAsync):
-            service_name = SERVICE_NAME
+            service_name = _SERVICE_NAME
 
             def __init__(
                 self,
@@ -821,7 +824,7 @@ async def list(
                 )
 
         class ProfileAsync(ServiceBaseAsync):
-            service_name = SERVICE_NAME
+            service_name = _SERVICE_NAME
 
             def __init__(
                 self,
@@ -905,8 +908,57 @@ async def update(
                     "v2/user/profile/update", m.UserProfileUpdateResult, data=input.model_dump(exclude_none=True)
                 )
 
+        class GroupAsync(ServiceBaseAsync):
+            service_name = _SERVICE_NAME
+
+            def __init__(
+                self,
+                token: str,
+                config: PangeaConfig | None = None,
+                logger_name: str = "pangea",
+            ) -> None:
+                super().__init__(token, config, logger_name=logger_name)
+
+            async def assign(self, user_id: str, group_ids: list[str]) -> PangeaResponse[PangeaResponseResult]:
+                """
+                Assign groups to a user
+
+                Add a list of groups to a specified user
+
+                OperationId: authn_post_v2_user_group_assign
+                """
+                return await self.request.post(
+                    "v2/user/group/assign",
+                    data={"id": user_id, "group_ids": group_ids},
+                    result_class=m.PangeaResponseResult,
+                )
+
+            async def remove(self, user_id: str, group_id: str) -> PangeaResponse[PangeaResponseResult]:
+                """
+                Remove a group assigned to a user
+
+                Remove a group assigned to a user
+
+                OperationId: authn_post_v2_user_group_remove
+                """
+                return await self.request.post(
+                    "v2/user/group/remove",
+                    data={"id": user_id, "group_id": group_id},
+                    result_class=m.PangeaResponseResult,
+                )
+
+            async def list(self, user_id: str) -> PangeaResponse[m.GroupList]:
+                """
+                List of groups assigned to a user
+
+                Return a list of ids for groups assigned to a user
+
+                OperationId: authn_post_v2_user_group_list
+                """
+                return await self.request.post("v2/user/group/list", data={"id": user_id}, result_class=m.GroupList)
+
     class FlowAsync(ServiceBaseAsync):
-        service_name = SERVICE_NAME
+        service_name = _SERVICE_NAME
 
         def __init__(
             self,
@@ -1052,7 +1104,7 @@ async def update(
             )
 
     class AgreementsAsync(ServiceBaseAsync):
-        service_name = SERVICE_NAME
+        service_name = _SERVICE_NAME
 
         def __init__(
             self,
@@ -1201,3 +1253,108 @@ async def update(
             return await self.request.post(
                 "v2/agreements/update", m.AgreementUpdateResult, data=input.model_dump(exclude_none=True)
             )
+
+    class GroupAsync(ServiceBaseAsync):
+        service_name = _SERVICE_NAME
+
+        def __init__(
+            self,
+            token: str,
+            config: PangeaConfig | None = None,
+            logger_name: str = "pangea",
+        ) -> None:
+            super().__init__(token, config, logger_name=logger_name)
+
+        async def create(
+            self, name: str, type: str, *, description: str | None = None, attributes: dict[str, str] | None = None
+        ) -> PangeaResponse[m.GroupInfo]:
+            """
+            Create a new group
+
+            Create a new group
+
+            OperationId: authn_post_v2_group_create
+            """
+            return await self.request.post(
+                "v2/group/create",
+                data={"name": name, "type": type, "description": description, "attributes": attributes},
+                result_class=m.GroupInfo,
+            )
+
+        async def delete(self, id: str) -> PangeaResponse[PangeaResponseResult]:
+            """
+            Delete a group
+
+            Delete a group
+
+            OperationId: authn_post_v2_group_delete
+            """
+            return await self.request.post("v2/group/delete", data={"id": id}, result_class=PangeaResponseResult)
+
+        async def get(self, id: str) -> PangeaResponse[m.GroupInfo]:
+            """
+            Get group information
+
+            Look up a group by ID and return its information.
+
+            OperationId: authn_post_v2_group_get
+            """
+            return await self.request.post("v2/group/get", data={"id": id}, result_class=m.GroupInfo)
+
+        async def list(
+            self,
+            *,
+            filter: m.GroupsFilter | None = None,
+            last: str | None = None,
+            order: Literal["asc", "desc"] | None = None,
+            order_by: Literal["id", "created_at", "updated_at", "name", "type"] | None = None,
+            size: int | None = None,
+        ) -> PangeaResponse[m.GroupList]:
+            """
+            List groups
+
+            Look up groups by name, type, or attributes.
+            """
+            return await self.request.post(
+                "v2/group/list",
+                data={"filter": filter, "last": last, "order": order, "order_by": order_by, "size": size},
+                result_class=m.GroupList,
+            )
+
+        async def list_users(
+            self, id: str, *, last: str | None = None, size: int | None = None
+        ) -> PangeaResponse[m.GroupUserList]:
+            """
+            List of users assigned to a group
+
+            Return a list of ids for users assigned to a group
+
+            OperationId: authn_post_v2_group_user_list
+            """
+            return await self.request.post(
+                "v2/group/user/list",
+                data={"id": id, "last": last, "size": size},
+                result_class=m.GroupUserList,
+            )
+
+        async def update(
+            self,
+            id: str,
+            *,
+            name: str | None = None,
+            description: str | None = None,
+            type: str | None = None,
+            attributes: dict[str, str] | None = None,
+        ) -> PangeaResponse[m.GroupInfo]:
+            """
+            Update group information
+
+            Update group information
+
+            OperationId: authn_post_v2_group_update
+            """
+            return await self.request.post(
+                "v2/group/update",
+                data={"id": id, "name": name, "description": description, "type": type, "attributes": attributes},
+                result_class=m.GroupInfo,
+            )