feat(iam): add logs listing method (#348)

Author: scaleway-bot

scaleway-async/scaleway_async/iam/v1alpha1/__init__.py

@@ -5,11 +5,14 @@
 from .types import ListApplicationsRequestOrderBy
 from .types import ListGroupsRequestOrderBy
 from .types import ListJWTsRequestOrderBy
+from .types import ListLogsRequestOrderBy
 from .types import ListPermissionSetsRequestOrderBy
 from .types import ListPoliciesRequestOrderBy
 from .types import ListQuotaRequestOrderBy
 from .types import ListSSHKeysRequestOrderBy
 from .types import ListUsersRequestOrderBy
+from .types import LogAction
+from .types import LogResourceType
 from .types import PermissionSetScopeType
 from .types import UserStatus
 from .types import UserType
@@ -21,12 +24,14 @@
 from .types import ListApplicationsResponse
 from .types import ListGroupsResponse
 from .types import ListJWTsResponse
+from .types import ListLogsResponse
 from .types import ListPermissionSetsResponse
 from .types import ListPoliciesResponse
 from .types import ListQuotaResponse
 from .types import ListRulesResponse
 from .types import ListSSHKeysResponse
 from .types import ListUsersResponse
+from .types import Log
 from .types import PermissionSet
 from .types import Policy
 from .types import Quotum
@@ -43,11 +48,14 @@
     "ListApplicationsRequestOrderBy",
     "ListGroupsRequestOrderBy",
     "ListJWTsRequestOrderBy",
+    "ListLogsRequestOrderBy",
     "ListPermissionSetsRequestOrderBy",
     "ListPoliciesRequestOrderBy",
     "ListQuotaRequestOrderBy",
     "ListSSHKeysRequestOrderBy",
     "ListUsersRequestOrderBy",
+    "LogAction",
+    "LogResourceType",
     "PermissionSetScopeType",
     "UserStatus",
     "UserType",
@@ -59,12 +67,14 @@
     "ListApplicationsResponse",
     "ListGroupsResponse",
     "ListJWTsResponse",
+    "ListLogsResponse",
     "ListPermissionSetsResponse",
     "ListPoliciesResponse",
     "ListQuotaResponse",
     "ListRulesResponse",
     "ListSSHKeysResponse",
     "ListUsersResponse",
+    "Log",
     "PermissionSet",
     "Policy",
     "Quotum",

scaleway-async/scaleway_async/iam/v1alpha1/api.py

@@ -18,11 +18,14 @@
     ListApplicationsRequestOrderBy,
     ListGroupsRequestOrderBy,
     ListJWTsRequestOrderBy,
+    ListLogsRequestOrderBy,
     ListPermissionSetsRequestOrderBy,
     ListPoliciesRequestOrderBy,
     ListQuotaRequestOrderBy,
     ListSSHKeysRequestOrderBy,
     ListUsersRequestOrderBy,
+    LogAction,
+    LogResourceType,
     APIKey,
     Application,
     Group,
@@ -31,12 +34,14 @@
     ListApplicationsResponse,
     ListGroupsResponse,
     ListJWTsResponse,
+    ListLogsResponse,
     ListPermissionSetsResponse,
     ListPoliciesResponse,
     ListQuotaResponse,
     ListRulesResponse,
     ListSSHKeysResponse,
     ListUsersResponse,
+    Log,
     PermissionSet,
     Policy,
     Quotum,
@@ -83,6 +88,7 @@
     unmarshal_Application,
     unmarshal_Group,
     unmarshal_JWT,
+    unmarshal_Log,
     unmarshal_Policy,
     unmarshal_Quotum,
     unmarshal_SSHKey,
@@ -91,6 +97,7 @@
     unmarshal_ListApplicationsResponse,
     unmarshal_ListGroupsResponse,
     unmarshal_ListJWTsResponse,
+    unmarshal_ListLogsResponse,
     unmarshal_ListPermissionSetsResponse,
     unmarshal_ListPoliciesResponse,
     unmarshal_ListQuotaResponse,
@@ -2102,3 +2109,106 @@ async def delete_jwt(
 
         self._throw_on_error(res)
         return None
+
+    async def list_logs(
+        self,
+        *,
+        order_by: ListLogsRequestOrderBy = ListLogsRequestOrderBy.CREATED_AT_ASC,
+        organization_id: Optional[str] = None,
+        page_size: Optional[int] = None,
+        page: Optional[int] = None,
+        created_after: Optional[datetime] = None,
+        created_before: Optional[datetime] = None,
+        action: LogAction = LogAction.UNKNOWN_ACTION,
+        resource_type: LogResourceType = LogResourceType.UNKNOWN_RESOURCE_TYPE,
+        search: Optional[str] = None,
+    ) -> ListLogsResponse:
+        """
+
+        Usage:
+        ::
+
+            result = await api.list_logs()
+        """
+
+        res = self._request(
+            "GET",
+            f"/iam/v1alpha1/logs",
+            params={
+                "action": action,
+                "created_after": created_after,
+                "created_before": created_before,
+                "order_by": order_by,
+                "organization_id": organization_id
+                or self.client.default_organization_id,
+                "page": page,
+                "page_size": page_size or self.client.default_page_size,
+                "resource_type": resource_type,
+                "search": search,
+            },
+        )
+
+        self._throw_on_error(res)
+        return unmarshal_ListLogsResponse(res.json())
+
+    async def list_logs_all(
+        self,
+        *,
+        order_by: Optional[ListLogsRequestOrderBy] = None,
+        organization_id: Optional[str] = None,
+        page_size: Optional[int] = None,
+        page: Optional[int] = None,
+        created_after: Optional[datetime] = None,
+        created_before: Optional[datetime] = None,
+        action: Optional[LogAction] = None,
+        resource_type: Optional[LogResourceType] = None,
+        search: Optional[str] = None,
+    ) -> List[Log]:
+        """
+        :return: :class:`List[ListLogsResponse] <List[ListLogsResponse]>`
+
+        Usage:
+        ::
+
+            result = await api.list_logs_all()
+        """
+
+        return await fetch_all_pages_async(
+            type=ListLogsResponse,
+            key="logs",
+            fetcher=self.list_logs,
+            args={
+                "order_by": order_by,
+                "organization_id": organization_id,
+                "page_size": page_size,
+                "page": page,
+                "created_after": created_after,
+                "created_before": created_before,
+                "action": action,
+                "resource_type": resource_type,
+                "search": search,
+            },
+        )
+
+    async def get_log(
+        self,
+        *,
+        log_id: str,
+    ) -> Log:
+        """
+
+        Usage:
+        ::
+
+            result = await api.get_log(log_id="example")
+        """
+
+        param_log_id = validate_path_param("log_id", log_id)
+
+        res = self._request(
+            "GET",
+            f"/iam/v1alpha1/logs/{param_log_id}",
+        )
+
+        self._throw_on_error(res)
+        return unmarshal_Log(res.json())

scaleway-async/scaleway_async/iam/v1alpha1/marshalling.py

@@ -18,12 +18,14 @@
     ListApplicationsResponse,
     ListGroupsResponse,
     ListJWTsResponse,
+    ListLogsResponse,
     ListPermissionSetsResponse,
     ListPoliciesResponse,
     ListQuotaResponse,
     ListRulesResponse,
     ListSSHKeysResponse,
     ListUsersResponse,
+    Log,
     PermissionSet,
     Policy,
     Quotum,
@@ -200,6 +202,44 @@ def unmarshal_JWT(data: Any) -> JWT:
     return JWT(**args)
 
 
+def unmarshal_Log(data: Any) -> Log:
+    if type(data) is not dict:
+        raise TypeError(
+            f"Unmarshalling the type 'Log' failed as data isn't a dictionary."
+        )
+
+    args: Dict[str, Any] = {}
+
+    field = data.get("action", None)
+    args["action"] = field
+
+    field = data.get("bearer_id", None)
+    args["bearer_id"] = field
+
+    field = data.get("created_at", None)
+    args["created_at"] = parser.isoparse(field) if type(field) is str else field
+
+    field = data.get("id", None)
+    args["id"] = field
+
+    field = data.get("ip", None)
+    args["ip"] = field
+
+    field = data.get("organization_id", None)
+    args["organization_id"] = field
+
+    field = data.get("resource_id", None)
+    args["resource_id"] = field
+
+    field = data.get("resource_type", None)
+    args["resource_type"] = field
+
+    field = data.get("user_agent", None)
+    args["user_agent"] = field
+
+    return Log(**args)
+
+
 def unmarshal_PermissionSet(data: Any) -> PermissionSet:
     if type(data) is not dict:
         raise TypeError(
@@ -485,6 +525,23 @@ def unmarshal_ListJWTsResponse(data: Any) -> ListJWTsResponse:
     return ListJWTsResponse(**args)
 
 
+def unmarshal_ListLogsResponse(data: Any) -> ListLogsResponse:
+    if type(data) is not dict:
+        raise TypeError(
+            f"Unmarshalling the type 'ListLogsResponse' failed as data isn't a dictionary."
+        )
+
+    args: Dict[str, Any] = {}
+
+    field = data.get("logs", None)
+    args["logs"] = [unmarshal_Log(v) for v in field] if field is not None else None
+
+    field = data.get("total_count", None)
+    args["total_count"] = field
+
+    return ListLogsResponse(**args)
+
+
 def unmarshal_ListPermissionSetsResponse(data: Any) -> ListPermissionSetsResponse:
     if type(data) is not dict:
         raise TypeError(