feat(client): add members related endpoints

Author: GG-Yanne

pygitguardian/client.py

@@ -26,6 +26,8 @@
 )
 from .models import (
     APITokensResponse,
+    CursorPaginatedResponse,
+    DeleteMember,
     Detail,
     Document,
     DocumentSchema,
@@ -34,13 +36,16 @@
     HoneytokenWithContextResponse,
     JWTResponse,
     JWTService,
+    Member,
+    MembersParameters,
     MultiScanResult,
     QuotaResponse,
     RemediationMessages,
     ScanResult,
     SecretIncident,
     SecretScanPreferences,
     ServerMetadata,
+    UpdateMember,
 )
 from .sca_models import (
     ComputeSCAFilesResult,
@@ -335,6 +340,40 @@ def post(
             **kwargs,
         )
 
+    def patch(
+        self,
+        endpoint: str,
+        data: Union[Dict[str, Any], List[Dict[str, Any]], None] = None,
+        version: str = DEFAULT_API_VERSION,
+        extra_headers: Optional[Dict[str, str]] = None,
+        **kwargs: Any,
+    ) -> Response:
+        return self.request(
+            "patch",
+            endpoint=endpoint,
+            json=data,
+            version=version,
+            extra_headers=extra_headers,
+            **kwargs,
+        )
+
+    def delete(
+        self,
+        endpoint: str,
+        data: Union[Dict[str, Any], List[Dict[str, Any]], None] = None,
+        version: str = DEFAULT_API_VERSION,
+        extra_headers: Optional[Dict[str, str]] = None,
+        **kwargs: Any,
+    ) -> Response:
+        return self.request(
+            "delete",
+            endpoint=endpoint,
+            json=data,
+            version=version,
+            extra_headers=extra_headers,
+            **kwargs,
+        )
+
     def health_check(self) -> HealthCheckResponse:
         """
         health_check handles the /health endpoint of the API
@@ -859,3 +898,84 @@ def scan_diff(
                 result = load_detail(response)
             result.status_code = response.status_code
         return result
+
+    def list_members(
+        self,
+        query_parameters: Optional[MembersParameters] = None,
+        extra_headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Detail, CursorPaginatedResponse[Member]]:
+
+        response = self.get(
+            endpoint="members",
+            data=query_parameters.to_dict() if query_parameters else {},
+            extra_headers=extra_headers,
+        )
+
+        obj: Union[Detail, CursorPaginatedResponse[Member]]
+        if is_ok(response):
+            obj = CursorPaginatedResponse[Member].from_response(response, Member)
+        else:
+            obj = load_detail(response)
+
+        obj.status_code = response.status_code
+        return obj
+
+    def get_member(
+        self,
+        member_id: int,
+        extra_headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Detail, Member]:
+        response = self.get(
+            endpoint=f"members/{member_id}",
+            extra_headers=extra_headers,
+        )
+        obj: Union[Detail, Member]
+        if is_ok(response):
+            obj = Member.from_dict(response.json())
+        else:
+            obj = load_detail(response)
+
+        obj.status_code = response.status_code
+        return obj
+
+    def update_member(
+        self,
+        payload: UpdateMember,
+        extra_headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Detail, Member]:
+
+        member_id = payload.id
+        data = UpdateMember.to_dict(payload)
+        del data["id"]
+
+        response = self.patch(
+            f"members/{member_id}", data=data, extra_headers=extra_headers
+        )
+        obj: Union[Detail, Member]
+        if is_ok(response):
+            obj = Member.from_dict(response.json())
+            print("Member : ", obj)
+        else:
+            obj = load_detail(response)
+
+        obj.status_code = response.status_code
+        return obj
+
+    def delete_member(
+        self,
+        member: DeleteMember,
+        extra_headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Detail, int]:
+        member_id = member.id
+        data = member.to_dict()
+        del data["id"]
+
+        response = self.delete(
+            f"members/{member_id}", params=data, extra_headers=extra_headers
+        )
+
+        # We bypass `is_ok` because the response content type is none
+        if response.status_code == 204:
+            return 204
+
+        return load_detail(response)

pygitguardian/models.py

@@ -4,7 +4,19 @@
 from dataclasses import dataclass, field
 from datetime import date, datetime
 from enum import Enum
-from typing import Any, ClassVar, Dict, List, Literal, Optional, Type, cast
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    ClassVar,
+    Dict,
+    Generic,
+    List,
+    Literal,
+    Optional,
+    Type,
+    TypeVar,
+    cast,
+)
 from uuid import UUID
 
 import marshmallow_dataclass
@@ -13,6 +25,7 @@
     Schema,
     ValidationError,
     fields,
+    post_dump,
     post_load,
     pre_load,
     validate,
@@ -28,6 +41,10 @@
 )
 
 
+if TYPE_CHECKING:
+    import requests
+
+
 class ToDictMixin:
     """
     Provides a type-safe `to_dict()` method for classes using Marshmallow
@@ -54,8 +71,8 @@ class FromDictMixin:
     SCHEMA: ClassVar[Schema]
 
     @classmethod
-    def from_dict(cls, dct: Dict[str, Any]) -> Self:
-        return cast(Self, cls.SCHEMA.load(dct))
+    def from_dict(cls, dct: Dict[str, Any], many: Optional[bool] = None) -> Self:
+        return cast(Self, cls.SCHEMA.load(dct, many=many))
 
 
 class BaseSchema(Schema):
@@ -1077,3 +1094,146 @@ def __repr__(self) -> str:
     marshmallow_dataclass.class_schema(SecretIncident, base_schema=BaseSchema),
 )
 SecretIncident.SCHEMA = SecretIncidentSchema()
+
+
+class AccessLevel(str, Enum):
+    OWNER = "owner"
+    MANAGER = "manager"
+    MEMBER = "member"
+    RESTRICTED = "restricted"
+
+
+class PaginationParameter(Base, FromDictMixin):
+    """Pagination mixin used for endpoints that support pagination."""
+
+    cursor: str = ""
+    per_page: int = 20
+
+
+class SearchParameter(Base, FromDictMixin):
+    search: Optional[str] = None
+
+
+PaginatedData = TypeVar("PaginatedData", bound=FromDictMixin)
+
+
+@dataclass
+class CursorPaginatedResponse(Generic[PaginatedData]):
+    status_code: int
+    data: list[PaginatedData]
+    prev: Optional[str] = None
+    next: Optional[str] = None
+
+    @classmethod
+    def from_response(
+        cls, response: "requests.Response", data_type: Type[PaginatedData]
+    ) -> "CursorPaginatedResponse[PaginatedData]":
+        data = cast(
+            list[PaginatedData], data_type.from_dict(response.json(), many=True)
+        )
+        paginated_response = cls(status_code=response.status_code, data=data)
+
+        if previous_page := response.links.get("prev"):
+            paginated_response.prev = previous_page["url"]
+        if next_page := response.links.get("next"):
+            paginated_response.prev = next_page["url"]
+
+        return paginated_response
+
+
+@dataclass
+class MembersParameters(PaginationParameter, SearchParameter, Base, FromDictMixin):
+    """
+    Members query parameters
+    """
+
+    access_level: Optional[AccessLevel] = None
+    active: Optional[bool] = None
+    ordering: Optional[
+        Literal["id", "-id", "created_at", "-created_at", "last_login", "-last_login"]
+    ] = None
+
+
+MembersParametersSchema = cast(
+    Type[BaseSchema],
+    marshmallow_dataclass.class_schema(MembersParameters, base_schema=BaseSchema),
+)
+MembersParameters.SCHEMA = MembersParametersSchema()
+
+
+@dataclass
+class Member(Base, FromDictMixin):
+    """
+    Member represents a user in a GitGuardian account.
+    """
+
+    id: int
+    access_level: AccessLevel
+    email: str
+    name: str
+    created_at: datetime
+    last_login: Optional[datetime]
+    active: bool
+
+
+class MemberSchema(BaseSchema):
+    id = fields.Int(required=True)
+    access_level = fields.Enum(AccessLevel, by_value=True, required=True)
+    email = fields.Str(required=True)
+    name = fields.Str(required=True)
+    created_at = fields.AwareDateTime(required=True)
+    last_login = fields.AwareDateTime(allow_none=True)
+    active = fields.Bool(required=True)
+
+    @post_load
+    def return_member(
+        self,
+        data: list[dict[str, Any]] | dict[str, Any],
+        **kwargs: dict[str, Any],
+    ):
+        data = cast(dict[str, Any], data)
+        return Member(**data)
+
+
+Member.SCHEMA = MemberSchema()
+
+
+class UpdateMemberSchema(BaseSchema):
+    id = fields.Int(required=True)
+    access_level = fields.Enum(AccessLevel, by_value=True, allow_none=True)
+    active = fields.Bool(allow_none=True)
+
+    @post_dump
+    def access_level_value(
+        self, data: dict[str, Any], **kwargs: dict[str, Any]
+    ) -> dict[str, Any]:
+        if "access_level" in data:
+            data["access_level"] = AccessLevel(data["access_level"]).value
+        return data
+
+
+@dataclass
+class UpdateMember(Base, FromDictMixin):
+    """
+    UpdateMember represnets the payload to update a member
+    """
+
+    id: int
+    access_level: Optional[AccessLevel] = None
+    active: Optional[bool] = None
+
+
+UpdateMember.SCHEMA = UpdateMemberSchema()
+
+
+@dataclass
+class DeleteMember(Base, FromDictMixin):
+    id: int
+    send_email: Optional[bool] = None
+
+
+DeleteMemberSchema = cast(
+    Type[BaseSchema],
+    marshmallow_dataclass.class_schema(DeleteMember, base_schema=BaseSchema),
+)
+DeleteMember.SCHEMA = DeleteMemberSchema()