feat(audit_trail): support system events (#1246)

Co-authored-by: Yacine Fodil <105779815+yfodil@users.noreply.github.com>

Author: scaleway-bot

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

@@ -6,8 +6,10 @@
 from .types import AuthenticationEventOrigin
 from .types import AuthenticationEventResult
 from .types import ListAuthenticationEventsRequestOrderBy
+from .types import ListCombinedEventsRequestOrderBy
 from .types import ListEventsRequestOrderBy
 from .types import ResourceType
+from .types import SystemEventKind
 from .types import AccountOrganizationInfo
 from .types import AccountProjectInfo
 from .types import AccountUserInfo
@@ -32,13 +34,16 @@
 from .types import SecretManagerSecretVersionInfo
 from .types import Resource
 from .types import EventPrincipal
-from .types import EventSystem
-from .types import ProductService
 from .types import AuthenticationEvent
 from .types import Event
+from .types import SystemEvent
+from .types import ProductService
+from .types import ListCombinedEventsResponseCombinedEvent
 from .types import Product
 from .types import ListAuthenticationEventsRequest
 from .types import ListAuthenticationEventsResponse
+from .types import ListCombinedEventsRequest
+from .types import ListCombinedEventsResponse
 from .types import ListEventsRequest
 from .types import ListEventsResponse
 from .types import ListProductsRequest
@@ -52,8 +57,10 @@
     "AuthenticationEventOrigin",
     "AuthenticationEventResult",
     "ListAuthenticationEventsRequestOrderBy",
+    "ListCombinedEventsRequestOrderBy",
     "ListEventsRequestOrderBy",
     "ResourceType",
+    "SystemEventKind",
     "AccountOrganizationInfo",
     "AccountProjectInfo",
     "AccountUserInfo",
@@ -78,13 +85,16 @@
     "SecretManagerSecretVersionInfo",
     "Resource",
     "EventPrincipal",
-    "EventSystem",
-    "ProductService",
     "AuthenticationEvent",
     "Event",
+    "SystemEvent",
+    "ProductService",
+    "ListCombinedEventsResponseCombinedEvent",
     "Product",
     "ListAuthenticationEventsRequest",
     "ListAuthenticationEventsResponse",
+    "ListCombinedEventsRequest",
+    "ListCombinedEventsResponse",
     "ListEventsRequest",
     "ListEventsResponse",
     "ListProductsRequest",

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

@@ -13,14 +13,17 @@
 )
 from .types import (
     ListAuthenticationEventsRequestOrderBy,
+    ListCombinedEventsRequestOrderBy,
     ListEventsRequestOrderBy,
     ResourceType,
     ListAuthenticationEventsResponse,
+    ListCombinedEventsResponse,
     ListEventsResponse,
     ListProductsResponse,
 )
 from .marshalling import (
     unmarshal_ListAuthenticationEventsResponse,
+    unmarshal_ListCombinedEventsResponse,
     unmarshal_ListEventsResponse,
     unmarshal_ListProductsResponse,
 )
@@ -158,6 +161,60 @@ async def list_authentication_events(
         self._throw_on_error(res)
         return unmarshal_ListAuthenticationEventsResponse(res.json())
 
+    async def list_combined_events(
+        self,
+        *,
+        region: Optional[ScwRegion] = None,
+        organization_id: Optional[str] = None,
+        project_id: Optional[str] = None,
+        resource_type: Optional[ResourceType] = None,
+        recorded_after: Optional[datetime] = None,
+        recorded_before: Optional[datetime] = None,
+        order_by: Optional[ListCombinedEventsRequestOrderBy] = None,
+        page_size: Optional[int] = None,
+        page_token: Optional[str] = None,
+    ) -> ListCombinedEventsResponse:
+        """
+        :param region: Region to target. If none is passed will use default region from the config.
+        :param organization_id:
+        :param project_id:
+        :param resource_type:
+        :param recorded_after:
+        :param recorded_before:
+        :param order_by:
+        :param page_size:
+        :param page_token:
+        :return: :class:`ListCombinedEventsResponse <ListCombinedEventsResponse>`
+
+        Usage:
+        ::
+
+            result = await api.list_combined_events()
+        """
+
+        param_region = validate_path_param(
+            "region", region or self.client.default_region
+        )
+
+        res = self._request(
+            "GET",
+            f"/audit-trail/v1alpha1/regions/{param_region}/combined-events",
+            params={
+                "order_by": order_by,
+                "organization_id": organization_id
+                or self.client.default_organization_id,
+                "page_size": page_size or self.client.default_page_size,
+                "page_token": page_token,
+                "project_id": project_id or self.client.default_project_id,
+                "recorded_after": recorded_after,
+                "recorded_before": recorded_before,
+                "resource_type": resource_type,
+            },
+        )
+
+        self._throw_on_error(res)
+        return unmarshal_ListCombinedEventsResponse(res.json())
+
     async def list_products(
         self,
         *,

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

@@ -36,8 +36,10 @@
     AuthenticationEvent,
     ListAuthenticationEventsResponse,
     EventPrincipal,
-    EventSystem,
     Event,
+    SystemEvent,
+    ListCombinedEventsResponseCombinedEvent,
+    ListCombinedEventsResponse,
     ListEventsResponse,
     ProductService,
     Product,
@@ -802,23 +804,6 @@ def unmarshal_EventPrincipal(data: Any) -> EventPrincipal:
     return EventPrincipal(**args)
 
 
-def unmarshal_EventSystem(data: Any) -> EventSystem:
-    if not isinstance(data, dict):
-        raise TypeError(
-            "Unmarshalling the type 'EventSystem' failed as data isn't a dictionary."
-        )
-
-    args: dict[str, Any] = {}
-
-    field = data.get("name", None)
-    if field is not None:
-        args["name"] = field
-    else:
-        args["name"] = None
-
-    return EventSystem(**args)
-
-
 def unmarshal_Event(data: Any) -> Event:
     if not isinstance(data, dict):
         raise TypeError(
@@ -857,18 +842,6 @@ def unmarshal_Event(data: Any) -> Event:
     else:
         args["product_name"] = None
 
-    field = data.get("service_name", None)
-    if field is not None:
-        args["service_name"] = field
-    else:
-        args["service_name"] = None
-
-    field = data.get("method_name", None)
-    if field is not None:
-        args["method_name"] = field
-    else:
-        args["method_name"] = None
-
     field = data.get("recorded_at", None)
     if field is not None:
         args["recorded_at"] = (
@@ -883,6 +856,30 @@ def unmarshal_Event(data: Any) -> Event:
     else:
         args["principal"] = None
 
+    field = data.get("project_id", None)
+    if field is not None:
+        args["project_id"] = field
+    else:
+        args["project_id"] = None
+
+    field = data.get("user_agent", None)
+    if field is not None:
+        args["user_agent"] = field
+    else:
+        args["user_agent"] = None
+
+    field = data.get("service_name", None)
+    if field is not None:
+        args["service_name"] = field
+    else:
+        args["service_name"] = None
+
+    field = data.get("method_name", None)
+    if field is not None:
+        args["method_name"] = field
+    else:
+        args["method_name"] = None
+
     field = data.get("resources", None)
     if field is not None:
         args["resources"] = (
@@ -903,31 +900,146 @@ def unmarshal_Event(data: Any) -> Event:
     else:
         args["status_code"] = 0
 
-    field = data.get("system", None)
+    field = data.get("request_body", None)
     if field is not None:
-        args["system"] = unmarshal_EventSystem(field)
+        args["request_body"] = field
     else:
-        args["system"] = None
+        args["request_body"] = {}
+
+    return Event(**args)
+
+
+def unmarshal_SystemEvent(data: Any) -> SystemEvent:
+    if not isinstance(data, dict):
+        raise TypeError(
+            "Unmarshalling the type 'SystemEvent' failed as data isn't a dictionary."
+        )
+
+    args: dict[str, Any] = {}
+
+    field = data.get("id", None)
+    if field is not None:
+        args["id"] = field
+    else:
+        args["id"] = None
+
+    field = data.get("locality", None)
+    if field is not None:
+        args["locality"] = field
+    else:
+        args["locality"] = None
+
+    field = data.get("organization_id", None)
+    if field is not None:
+        args["organization_id"] = field
+    else:
+        args["organization_id"] = None
+
+    field = data.get("source", None)
+    if field is not None:
+        args["source"] = field
+    else:
+        args["source"] = None
+
+    field = data.get("system_name", None)
+    if field is not None:
+        args["system_name"] = field
+    else:
+        args["system_name"] = None
+
+    field = data.get("resources", None)
+    if field is not None:
+        args["resources"] = (
+            [unmarshal_Resource(v) for v in field] if field is not None else None
+        )
+    else:
+        args["resources"] = None
+
+    field = data.get("kind", None)
+    if field is not None:
+        args["kind"] = field
+    else:
+        args["kind"] = None
+
+    field = data.get("product_name", None)
+    if field is not None:
+        args["product_name"] = field
+    else:
+        args["product_name"] = None
+
+    field = data.get("recorded_at", None)
+    if field is not None:
+        args["recorded_at"] = (
+            parser.isoparse(field) if isinstance(field, str) else field
+        )
+    else:
+        args["recorded_at"] = None
 
     field = data.get("project_id", None)
     if field is not None:
         args["project_id"] = field
     else:
         args["project_id"] = None
 
-    field = data.get("user_agent", None)
+    return SystemEvent(**args)
+
+
+def unmarshal_ListCombinedEventsResponseCombinedEvent(
+    data: Any,
+) -> ListCombinedEventsResponseCombinedEvent:
+    if not isinstance(data, dict):
+        raise TypeError(
+            "Unmarshalling the type 'ListCombinedEventsResponseCombinedEvent' failed as data isn't a dictionary."
+        )
+
+    args: dict[str, Any] = {}
+
+    field = data.get("api", None)
     if field is not None:
-        args["user_agent"] = field
+        args["api"] = unmarshal_Event(field)
     else:
-        args["user_agent"] = None
+        args["api"] = None
 
-    field = data.get("request_body", None)
+    field = data.get("auth", None)
     if field is not None:
-        args["request_body"] = field
+        args["auth"] = unmarshal_AuthenticationEvent(field)
     else:
-        args["request_body"] = {}
+        args["auth"] = None
 
-    return Event(**args)
+    field = data.get("system", None)
+    if field is not None:
+        args["system"] = unmarshal_SystemEvent(field)
+    else:
+        args["system"] = None
+
+    return ListCombinedEventsResponseCombinedEvent(**args)
+
+
+def unmarshal_ListCombinedEventsResponse(data: Any) -> ListCombinedEventsResponse:
+    if not isinstance(data, dict):
+        raise TypeError(
+            "Unmarshalling the type 'ListCombinedEventsResponse' failed as data isn't a dictionary."
+        )
+
+    args: dict[str, Any] = {}
+
+    field = data.get("events", None)
+    if field is not None:
+        args["events"] = (
+            [unmarshal_ListCombinedEventsResponseCombinedEvent(v) for v in field]
+            if field is not None
+            else None
+        )
+    else:
+        args["events"] = None
+
+    field = data.get("next_page_token", None)
+    if field is not None:
+        args["next_page_token"] = field
+    else:
+        args["next_page_token"] = None
+
+    return ListCombinedEventsResponse(**args)
 
 
 def unmarshal_ListEventsResponse(data: Any) -> ListEventsResponse: