diff --git a/scaleway-async/scaleway_async/audit_trail/__init__.py b/scaleway-async/scaleway_async/audit_trail/__init__.py new file mode 100644 index 000000000..8b74a5ed7 --- /dev/null +++ b/scaleway-async/scaleway_async/audit_trail/__init__.py @@ -0,0 +1,2 @@ +# This file was automatically generated. DO NOT EDIT. +# If you have any remark or suggestion do not hesitate to open an issue. diff --git a/scaleway-async/scaleway_async/audit_trail/v1alpha1/__init__.py b/scaleway-async/scaleway_async/audit_trail/v1alpha1/__init__.py new file mode 100644 index 000000000..1b89c7430 --- /dev/null +++ b/scaleway-async/scaleway_async/audit_trail/v1alpha1/__init__.py @@ -0,0 +1,37 @@ +# This file was automatically generated. DO NOT EDIT. +# If you have any remark or suggestion do not hesitate to open an issue. +from .types import ListEventsRequestOrderBy +from .types import ResourceType +from .types import KubernetesClusterInfo +from .types import KubernetesNodeInfo +from .types import KubernetesPoolInfo +from .types import SecretManagerSecretInfo +from .types import SecretManagerSecretVersionInfo +from .types import EventPrincipal +from .types import Resource +from .types import Event +from .types import Product +from .types import ListEventsRequest +from .types import ListEventsResponse +from .types import ListProductsRequest +from .types import ListProductsResponse +from .api import AuditTrailV1Alpha1API + +__all__ = [ + "ListEventsRequestOrderBy", + "ResourceType", + "KubernetesClusterInfo", + "KubernetesNodeInfo", + "KubernetesPoolInfo", + "SecretManagerSecretInfo", + "SecretManagerSecretVersionInfo", + "EventPrincipal", + "Resource", + "Event", + "Product", + "ListEventsRequest", + "ListEventsResponse", + "ListProductsRequest", + "ListProductsResponse", + "AuditTrailV1Alpha1API", +] diff --git a/scaleway-async/scaleway_async/audit_trail/v1alpha1/api.py b/scaleway-async/scaleway_async/audit_trail/v1alpha1/api.py new file mode 100644 index 000000000..9d31ea60e --- /dev/null +++ b/scaleway-async/scaleway_async/audit_trail/v1alpha1/api.py @@ -0,0 +1,122 @@ +# This file was automatically generated. DO NOT EDIT. +# If you have any remark or suggestion do not hesitate to open an issue. + +from datetime import datetime +from typing import Optional + +from scaleway_core.api import API +from scaleway_core.bridge import ( + Region, +) +from scaleway_core.utils import ( + validate_path_param, +) +from .types import ( + ListEventsRequestOrderBy, + ResourceType, + ListEventsResponse, + ListProductsResponse, +) +from .marshalling import ( + unmarshal_ListEventsResponse, + unmarshal_ListProductsResponse, +) + + +class AuditTrailV1Alpha1API(API): + """ + This API allows you to ensure accountability and security by recording events and changes performed within your Scaleway Organization. + """ + + async def list_events( + self, + *, + region: Optional[Region] = None, + project_id: Optional[str] = None, + organization_id: Optional[str] = None, + resource_type: Optional[ResourceType] = None, + method_name: Optional[str] = None, + status: Optional[int] = None, + recorded_after: Optional[datetime] = None, + recorded_before: Optional[datetime] = None, + order_by: Optional[ListEventsRequestOrderBy] = None, + page_size: Optional[int] = None, + page_token: Optional[str] = None, + product_name: Optional[str] = None, + ) -> ListEventsResponse: + """ + List events. + Retrieve the list of Audit Trail events for a Scaleway Organization and/or Project. You must specify the `organization_id` and optionally, the `project_id`. + :param region: Region to target. If none is passed will use default region from the config. + :param project_id: (Optional) ID of the Project containing the Audit Trail events. + :param organization_id: ID of the Organization containing the Audit Trail events. + :param resource_type: (Optional) Returns a paginated list of Scaleway resources' features. + :param method_name: (Optional) Name of the method or the API call performed. + :param status: (Optional) HTTP status code of the request. Returns either `200` if the request was successful or `403` if the permission was denied. + :param recorded_after: (Optional) The `recorded_after` parameter defines the earliest timestamp from which Audit Trail events are retrieved. Returns `one hour ago` by default. + :param recorded_before: (Optional) The `recorded_before` parameter defines the latest timestamp up to which Audit Trail events are retrieved. Returns `now` by default. + :param order_by: + :param page_size: + :param page_token: + :param product_name: (Optional) Name of the Scaleway resource in a hyphenated format. + :return: :class:`ListEventsResponse ` + + Usage: + :: + + result = await api.list_events() + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "GET", + f"/audit-trail/v1alpha1/regions/{param_region}/events", + params={ + "method_name": method_name, + "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, + "product_name": product_name, + "project_id": project_id or self.client.default_project_id, + "recorded_after": recorded_after, + "recorded_before": recorded_before, + "resource_type": resource_type, + "status": status, + }, + ) + + self._throw_on_error(res) + return unmarshal_ListEventsResponse(res.json()) + + async def list_products( + self, + *, + region: Optional[Region] = None, + ) -> ListProductsResponse: + """ + Retrieve the list of Scaleway resources for which you have Audit Trail events. + :param region: Region to target. If none is passed will use default region from the config. + :return: :class:`ListProductsResponse ` + + Usage: + :: + + result = await api.list_products() + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "GET", + f"/audit-trail/v1alpha1/regions/{param_region}/products", + ) + + self._throw_on_error(res) + return unmarshal_ListProductsResponse(res.json()) diff --git a/scaleway-async/scaleway_async/audit_trail/v1alpha1/marshalling.py b/scaleway-async/scaleway_async/audit_trail/v1alpha1/marshalling.py new file mode 100644 index 000000000..9223aea78 --- /dev/null +++ b/scaleway-async/scaleway_async/audit_trail/v1alpha1/marshalling.py @@ -0,0 +1,338 @@ +# This file was automatically generated. DO NOT EDIT. +# If you have any remark or suggestion do not hesitate to open an issue. + +from typing import Any, Dict +from dateutil import parser + +from .types import ( + KubernetesClusterInfo, + KubernetesNodeInfo, + KubernetesPoolInfo, + SecretManagerSecretInfo, + SecretManagerSecretVersionInfo, + EventPrincipal, + Resource, + Event, + ListEventsResponse, + Product, + ListProductsResponse, +) + + +def unmarshal_KubernetesClusterInfo(data: Any) -> KubernetesClusterInfo: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'KubernetesClusterInfo' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + return KubernetesClusterInfo(**args) + + +def unmarshal_KubernetesNodeInfo(data: Any) -> KubernetesNodeInfo: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'KubernetesNodeInfo' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("id", None) + if field is not None: + args["id"] = field + + field = data.get("name", None) + if field is not None: + args["name"] = field + + return KubernetesNodeInfo(**args) + + +def unmarshal_KubernetesPoolInfo(data: Any) -> KubernetesPoolInfo: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'KubernetesPoolInfo' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("id", None) + if field is not None: + args["id"] = field + + field = data.get("name", None) + if field is not None: + args["name"] = field + + return KubernetesPoolInfo(**args) + + +def unmarshal_SecretManagerSecretInfo(data: Any) -> SecretManagerSecretInfo: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'SecretManagerSecretInfo' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("path", None) + if field is not None: + args["path"] = field + + return SecretManagerSecretInfo(**args) + + +def unmarshal_SecretManagerSecretVersionInfo( + data: Any, +) -> SecretManagerSecretVersionInfo: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'SecretManagerSecretVersionInfo' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("revision", None) + if field is not None: + args["revision"] = field + + return SecretManagerSecretVersionInfo(**args) + + +def unmarshal_EventPrincipal(data: Any) -> EventPrincipal: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'EventPrincipal' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("id", None) + if field is not None: + args["id"] = field + + return EventPrincipal(**args) + + +def unmarshal_Resource(data: Any) -> Resource: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'Resource' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("id", None) + if field is not None: + args["id"] = field + + field = data.get("type", None) + if field is not None: + args["type_"] = field + + field = data.get("created_at", None) + if field is not None: + args["created_at"] = parser.isoparse(field) if isinstance(field, str) else field + else: + args["created_at"] = None + + field = data.get("updated_at", None) + if field is not None: + args["updated_at"] = parser.isoparse(field) if isinstance(field, str) else field + else: + args["updated_at"] = None + + field = data.get("deleted_at", None) + if field is not None: + args["deleted_at"] = parser.isoparse(field) if isinstance(field, str) else field + else: + args["deleted_at"] = None + + field = data.get("name", None) + if field is not None: + args["name"] = field + else: + args["name"] = None + + field = data.get("secm_secret_info", None) + if field is not None: + args["secm_secret_info"] = unmarshal_SecretManagerSecretInfo(field) + else: + args["secm_secret_info"] = None + + field = data.get("secm_secret_version_info", None) + if field is not None: + args["secm_secret_version_info"] = unmarshal_SecretManagerSecretVersionInfo( + field + ) + else: + args["secm_secret_version_info"] = None + + field = data.get("kube_cluster_info", None) + if field is not None: + args["kube_cluster_info"] = unmarshal_KubernetesClusterInfo(field) + else: + args["kube_cluster_info"] = None + + field = data.get("kube_pool_info", None) + if field is not None: + args["kube_pool_info"] = unmarshal_KubernetesPoolInfo(field) + else: + args["kube_pool_info"] = None + + field = data.get("kube_node_info", None) + if field is not None: + args["kube_node_info"] = unmarshal_KubernetesNodeInfo(field) + else: + args["kube_node_info"] = None + + return Resource(**args) + + +def unmarshal_Event(data: Any) -> Event: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'Event' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("id", None) + if field is not None: + args["id"] = field + + field = data.get("locality", None) + if field is not None: + args["locality"] = field + + field = data.get("organization_id", None) + if field is not None: + args["organization_id"] = field + + field = data.get("source_ip", None) + if field is not None: + args["source_ip"] = field + + field = data.get("product_name", None) + if field is not None: + args["product_name"] = field + + 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("principal", None) + if field is not None: + args["principal"] = unmarshal_EventPrincipal(field) + 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 + + field = data.get("method_name", None) + if field is not None: + args["method_name"] = field + + field = data.get("request_id", None) + if field is not None: + args["request_id"] = field + + field = data.get("status_code", None) + if field is not None: + args["status_code"] = field + + field = data.get("resource", None) + if field is not None: + args["resource"] = unmarshal_Resource(field) + else: + args["resource"] = None + + field = data.get("request_body", None) + if field is not None: + args["request_body"] = field + else: + args["request_body"] = None + + return Event(**args) + + +def unmarshal_ListEventsResponse(data: Any) -> ListEventsResponse: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'ListEventsResponse' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("events", None) + if field is not None: + args["events"] = ( + [unmarshal_Event(v) for v in field] if field is not None else 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 ListEventsResponse(**args) + + +def unmarshal_Product(data: Any) -> Product: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'Product' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("title", None) + if field is not None: + args["title"] = field + + field = data.get("name", None) + if field is not None: + args["name"] = field + + return Product(**args) + + +def unmarshal_ListProductsResponse(data: Any) -> ListProductsResponse: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'ListProductsResponse' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("products", None) + if field is not None: + args["products"] = ( + [unmarshal_Product(v) for v in field] if field is not None else None + ) + + field = data.get("total_count", None) + if field is not None: + args["total_count"] = field + + return ListProductsResponse(**args) diff --git a/scaleway-async/scaleway_async/audit_trail/v1alpha1/types.py b/scaleway-async/scaleway_async/audit_trail/v1alpha1/types.py new file mode 100644 index 000000000..2d9381701 --- /dev/null +++ b/scaleway-async/scaleway_async/audit_trail/v1alpha1/types.py @@ -0,0 +1,216 @@ +# This file was automatically generated. DO NOT EDIT. +# If you have any remark or suggestion do not hesitate to open an issue. +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from enum import Enum +from typing import Any, Dict, List, Optional + +from scaleway_core.bridge import ( + Region, +) +from scaleway_core.utils import ( + StrEnumMeta, +) + + +class ListEventsRequestOrderBy(str, Enum, metaclass=StrEnumMeta): + RECORDED_AT_DESC = "recorded_at_desc" + RECORDED_AT_ASC = "recorded_at_asc" + + def __str__(self) -> str: + return str(self.value) + + +class ResourceType(str, Enum, metaclass=StrEnumMeta): + UNKNOWN_TYPE = "unknown_type" + SECM_SECRET = "secm_secret" + SECM_SECRET_VERSION = "secm_secret_version" + KUBE_CLUSTER = "kube_cluster" + KUBE_POOL = "kube_pool" + KUBE_NODE = "kube_node" + + def __str__(self) -> str: + return str(self.value) + + +@dataclass +class KubernetesClusterInfo: + pass + + +@dataclass +class KubernetesNodeInfo: + id: str + + name: str + + +@dataclass +class KubernetesPoolInfo: + id: str + + name: str + + +@dataclass +class SecretManagerSecretInfo: + path: str + + +@dataclass +class SecretManagerSecretVersionInfo: + revision: int + + +@dataclass +class EventPrincipal: + id: str + + +@dataclass +class Resource: + id: str + + type_: ResourceType + + created_at: Optional[datetime] + + updated_at: Optional[datetime] + + deleted_at: Optional[datetime] + + name: Optional[str] + + secm_secret_info: Optional[SecretManagerSecretInfo] + + secm_secret_version_info: Optional[SecretManagerSecretVersionInfo] + + kube_cluster_info: Optional[KubernetesClusterInfo] + + kube_pool_info: Optional[KubernetesPoolInfo] + + kube_node_info: Optional[KubernetesNodeInfo] + + +@dataclass +class Event: + id: str + + locality: str + + organization_id: str + + source_ip: str + + product_name: str + + recorded_at: Optional[datetime] + + principal: Optional[EventPrincipal] + + project_id: Optional[str] + + user_agent: Optional[str] + + service_name: str + + method_name: str + + request_id: str + + status_code: int + + resource: Optional[Resource] + + request_body: Optional[Dict[str, Any]] + + +@dataclass +class Product: + title: str + + name: str + + +@dataclass +class ListEventsRequest: + region: Optional[Region] + """ + Region to target. If none is passed will use default region from the config. + """ + + project_id: Optional[str] + """ + (Optional) ID of the Project containing the Audit Trail events. + """ + + organization_id: Optional[str] + """ + ID of the Organization containing the Audit Trail events. + """ + + resource_type: Optional[ResourceType] + """ + (Optional) Returns a paginated list of Scaleway resources' features. + """ + + method_name: Optional[str] + """ + (Optional) Name of the method or the API call performed. + """ + + status: Optional[int] + """ + (Optional) HTTP status code of the request. Returns either `200` if the request was successful or `403` if the permission was denied. + """ + + recorded_after: Optional[datetime] + """ + (Optional) The `recorded_after` parameter defines the earliest timestamp from which Audit Trail events are retrieved. Returns `one hour ago` by default. + """ + + recorded_before: Optional[datetime] + """ + (Optional) The `recorded_before` parameter defines the latest timestamp up to which Audit Trail events are retrieved. Returns `now` by default. + """ + + order_by: Optional[ListEventsRequestOrderBy] + + page_size: Optional[int] + + page_token: Optional[str] + + product_name: Optional[str] + """ + (Optional) Name of the Scaleway resource in a hyphenated format. + """ + + +@dataclass +class ListEventsResponse: + events: List[Event] + """ + Single page of events matching the requested criteria. + """ + + next_page_token: Optional[str] + """ + Page token to use in following calls to keep listing. + """ + + +@dataclass +class ListProductsRequest: + region: Optional[Region] + """ + Region to target. If none is passed will use default region from the config. + """ + + +@dataclass +class ListProductsResponse: + products: List[Product] + + total_count: int diff --git a/scaleway/scaleway/audit_trail/__init__.py b/scaleway/scaleway/audit_trail/__init__.py new file mode 100644 index 000000000..8b74a5ed7 --- /dev/null +++ b/scaleway/scaleway/audit_trail/__init__.py @@ -0,0 +1,2 @@ +# This file was automatically generated. DO NOT EDIT. +# If you have any remark or suggestion do not hesitate to open an issue. diff --git a/scaleway/scaleway/audit_trail/v1alpha1/__init__.py b/scaleway/scaleway/audit_trail/v1alpha1/__init__.py new file mode 100644 index 000000000..1b89c7430 --- /dev/null +++ b/scaleway/scaleway/audit_trail/v1alpha1/__init__.py @@ -0,0 +1,37 @@ +# This file was automatically generated. DO NOT EDIT. +# If you have any remark or suggestion do not hesitate to open an issue. +from .types import ListEventsRequestOrderBy +from .types import ResourceType +from .types import KubernetesClusterInfo +from .types import KubernetesNodeInfo +from .types import KubernetesPoolInfo +from .types import SecretManagerSecretInfo +from .types import SecretManagerSecretVersionInfo +from .types import EventPrincipal +from .types import Resource +from .types import Event +from .types import Product +from .types import ListEventsRequest +from .types import ListEventsResponse +from .types import ListProductsRequest +from .types import ListProductsResponse +from .api import AuditTrailV1Alpha1API + +__all__ = [ + "ListEventsRequestOrderBy", + "ResourceType", + "KubernetesClusterInfo", + "KubernetesNodeInfo", + "KubernetesPoolInfo", + "SecretManagerSecretInfo", + "SecretManagerSecretVersionInfo", + "EventPrincipal", + "Resource", + "Event", + "Product", + "ListEventsRequest", + "ListEventsResponse", + "ListProductsRequest", + "ListProductsResponse", + "AuditTrailV1Alpha1API", +] diff --git a/scaleway/scaleway/audit_trail/v1alpha1/api.py b/scaleway/scaleway/audit_trail/v1alpha1/api.py new file mode 100644 index 000000000..7f31f84e2 --- /dev/null +++ b/scaleway/scaleway/audit_trail/v1alpha1/api.py @@ -0,0 +1,122 @@ +# This file was automatically generated. DO NOT EDIT. +# If you have any remark or suggestion do not hesitate to open an issue. + +from datetime import datetime +from typing import Optional + +from scaleway_core.api import API +from scaleway_core.bridge import ( + Region, +) +from scaleway_core.utils import ( + validate_path_param, +) +from .types import ( + ListEventsRequestOrderBy, + ResourceType, + ListEventsResponse, + ListProductsResponse, +) +from .marshalling import ( + unmarshal_ListEventsResponse, + unmarshal_ListProductsResponse, +) + + +class AuditTrailV1Alpha1API(API): + """ + This API allows you to ensure accountability and security by recording events and changes performed within your Scaleway Organization. + """ + + def list_events( + self, + *, + region: Optional[Region] = None, + project_id: Optional[str] = None, + organization_id: Optional[str] = None, + resource_type: Optional[ResourceType] = None, + method_name: Optional[str] = None, + status: Optional[int] = None, + recorded_after: Optional[datetime] = None, + recorded_before: Optional[datetime] = None, + order_by: Optional[ListEventsRequestOrderBy] = None, + page_size: Optional[int] = None, + page_token: Optional[str] = None, + product_name: Optional[str] = None, + ) -> ListEventsResponse: + """ + List events. + Retrieve the list of Audit Trail events for a Scaleway Organization and/or Project. You must specify the `organization_id` and optionally, the `project_id`. + :param region: Region to target. If none is passed will use default region from the config. + :param project_id: (Optional) ID of the Project containing the Audit Trail events. + :param organization_id: ID of the Organization containing the Audit Trail events. + :param resource_type: (Optional) Returns a paginated list of Scaleway resources' features. + :param method_name: (Optional) Name of the method or the API call performed. + :param status: (Optional) HTTP status code of the request. Returns either `200` if the request was successful or `403` if the permission was denied. + :param recorded_after: (Optional) The `recorded_after` parameter defines the earliest timestamp from which Audit Trail events are retrieved. Returns `one hour ago` by default. + :param recorded_before: (Optional) The `recorded_before` parameter defines the latest timestamp up to which Audit Trail events are retrieved. Returns `now` by default. + :param order_by: + :param page_size: + :param page_token: + :param product_name: (Optional) Name of the Scaleway resource in a hyphenated format. + :return: :class:`ListEventsResponse ` + + Usage: + :: + + result = api.list_events() + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "GET", + f"/audit-trail/v1alpha1/regions/{param_region}/events", + params={ + "method_name": method_name, + "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, + "product_name": product_name, + "project_id": project_id or self.client.default_project_id, + "recorded_after": recorded_after, + "recorded_before": recorded_before, + "resource_type": resource_type, + "status": status, + }, + ) + + self._throw_on_error(res) + return unmarshal_ListEventsResponse(res.json()) + + def list_products( + self, + *, + region: Optional[Region] = None, + ) -> ListProductsResponse: + """ + Retrieve the list of Scaleway resources for which you have Audit Trail events. + :param region: Region to target. If none is passed will use default region from the config. + :return: :class:`ListProductsResponse ` + + Usage: + :: + + result = api.list_products() + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "GET", + f"/audit-trail/v1alpha1/regions/{param_region}/products", + ) + + self._throw_on_error(res) + return unmarshal_ListProductsResponse(res.json()) diff --git a/scaleway/scaleway/audit_trail/v1alpha1/marshalling.py b/scaleway/scaleway/audit_trail/v1alpha1/marshalling.py new file mode 100644 index 000000000..9223aea78 --- /dev/null +++ b/scaleway/scaleway/audit_trail/v1alpha1/marshalling.py @@ -0,0 +1,338 @@ +# This file was automatically generated. DO NOT EDIT. +# If you have any remark or suggestion do not hesitate to open an issue. + +from typing import Any, Dict +from dateutil import parser + +from .types import ( + KubernetesClusterInfo, + KubernetesNodeInfo, + KubernetesPoolInfo, + SecretManagerSecretInfo, + SecretManagerSecretVersionInfo, + EventPrincipal, + Resource, + Event, + ListEventsResponse, + Product, + ListProductsResponse, +) + + +def unmarshal_KubernetesClusterInfo(data: Any) -> KubernetesClusterInfo: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'KubernetesClusterInfo' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + return KubernetesClusterInfo(**args) + + +def unmarshal_KubernetesNodeInfo(data: Any) -> KubernetesNodeInfo: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'KubernetesNodeInfo' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("id", None) + if field is not None: + args["id"] = field + + field = data.get("name", None) + if field is not None: + args["name"] = field + + return KubernetesNodeInfo(**args) + + +def unmarshal_KubernetesPoolInfo(data: Any) -> KubernetesPoolInfo: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'KubernetesPoolInfo' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("id", None) + if field is not None: + args["id"] = field + + field = data.get("name", None) + if field is not None: + args["name"] = field + + return KubernetesPoolInfo(**args) + + +def unmarshal_SecretManagerSecretInfo(data: Any) -> SecretManagerSecretInfo: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'SecretManagerSecretInfo' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("path", None) + if field is not None: + args["path"] = field + + return SecretManagerSecretInfo(**args) + + +def unmarshal_SecretManagerSecretVersionInfo( + data: Any, +) -> SecretManagerSecretVersionInfo: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'SecretManagerSecretVersionInfo' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("revision", None) + if field is not None: + args["revision"] = field + + return SecretManagerSecretVersionInfo(**args) + + +def unmarshal_EventPrincipal(data: Any) -> EventPrincipal: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'EventPrincipal' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("id", None) + if field is not None: + args["id"] = field + + return EventPrincipal(**args) + + +def unmarshal_Resource(data: Any) -> Resource: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'Resource' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("id", None) + if field is not None: + args["id"] = field + + field = data.get("type", None) + if field is not None: + args["type_"] = field + + field = data.get("created_at", None) + if field is not None: + args["created_at"] = parser.isoparse(field) if isinstance(field, str) else field + else: + args["created_at"] = None + + field = data.get("updated_at", None) + if field is not None: + args["updated_at"] = parser.isoparse(field) if isinstance(field, str) else field + else: + args["updated_at"] = None + + field = data.get("deleted_at", None) + if field is not None: + args["deleted_at"] = parser.isoparse(field) if isinstance(field, str) else field + else: + args["deleted_at"] = None + + field = data.get("name", None) + if field is not None: + args["name"] = field + else: + args["name"] = None + + field = data.get("secm_secret_info", None) + if field is not None: + args["secm_secret_info"] = unmarshal_SecretManagerSecretInfo(field) + else: + args["secm_secret_info"] = None + + field = data.get("secm_secret_version_info", None) + if field is not None: + args["secm_secret_version_info"] = unmarshal_SecretManagerSecretVersionInfo( + field + ) + else: + args["secm_secret_version_info"] = None + + field = data.get("kube_cluster_info", None) + if field is not None: + args["kube_cluster_info"] = unmarshal_KubernetesClusterInfo(field) + else: + args["kube_cluster_info"] = None + + field = data.get("kube_pool_info", None) + if field is not None: + args["kube_pool_info"] = unmarshal_KubernetesPoolInfo(field) + else: + args["kube_pool_info"] = None + + field = data.get("kube_node_info", None) + if field is not None: + args["kube_node_info"] = unmarshal_KubernetesNodeInfo(field) + else: + args["kube_node_info"] = None + + return Resource(**args) + + +def unmarshal_Event(data: Any) -> Event: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'Event' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("id", None) + if field is not None: + args["id"] = field + + field = data.get("locality", None) + if field is not None: + args["locality"] = field + + field = data.get("organization_id", None) + if field is not None: + args["organization_id"] = field + + field = data.get("source_ip", None) + if field is not None: + args["source_ip"] = field + + field = data.get("product_name", None) + if field is not None: + args["product_name"] = field + + 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("principal", None) + if field is not None: + args["principal"] = unmarshal_EventPrincipal(field) + 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 + + field = data.get("method_name", None) + if field is not None: + args["method_name"] = field + + field = data.get("request_id", None) + if field is not None: + args["request_id"] = field + + field = data.get("status_code", None) + if field is not None: + args["status_code"] = field + + field = data.get("resource", None) + if field is not None: + args["resource"] = unmarshal_Resource(field) + else: + args["resource"] = None + + field = data.get("request_body", None) + if field is not None: + args["request_body"] = field + else: + args["request_body"] = None + + return Event(**args) + + +def unmarshal_ListEventsResponse(data: Any) -> ListEventsResponse: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'ListEventsResponse' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("events", None) + if field is not None: + args["events"] = ( + [unmarshal_Event(v) for v in field] if field is not None else 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 ListEventsResponse(**args) + + +def unmarshal_Product(data: Any) -> Product: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'Product' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("title", None) + if field is not None: + args["title"] = field + + field = data.get("name", None) + if field is not None: + args["name"] = field + + return Product(**args) + + +def unmarshal_ListProductsResponse(data: Any) -> ListProductsResponse: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'ListProductsResponse' failed as data isn't a dictionary." + ) + + args: Dict[str, Any] = {} + + field = data.get("products", None) + if field is not None: + args["products"] = ( + [unmarshal_Product(v) for v in field] if field is not None else None + ) + + field = data.get("total_count", None) + if field is not None: + args["total_count"] = field + + return ListProductsResponse(**args) diff --git a/scaleway/scaleway/audit_trail/v1alpha1/types.py b/scaleway/scaleway/audit_trail/v1alpha1/types.py new file mode 100644 index 000000000..2d9381701 --- /dev/null +++ b/scaleway/scaleway/audit_trail/v1alpha1/types.py @@ -0,0 +1,216 @@ +# This file was automatically generated. DO NOT EDIT. +# If you have any remark or suggestion do not hesitate to open an issue. +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from enum import Enum +from typing import Any, Dict, List, Optional + +from scaleway_core.bridge import ( + Region, +) +from scaleway_core.utils import ( + StrEnumMeta, +) + + +class ListEventsRequestOrderBy(str, Enum, metaclass=StrEnumMeta): + RECORDED_AT_DESC = "recorded_at_desc" + RECORDED_AT_ASC = "recorded_at_asc" + + def __str__(self) -> str: + return str(self.value) + + +class ResourceType(str, Enum, metaclass=StrEnumMeta): + UNKNOWN_TYPE = "unknown_type" + SECM_SECRET = "secm_secret" + SECM_SECRET_VERSION = "secm_secret_version" + KUBE_CLUSTER = "kube_cluster" + KUBE_POOL = "kube_pool" + KUBE_NODE = "kube_node" + + def __str__(self) -> str: + return str(self.value) + + +@dataclass +class KubernetesClusterInfo: + pass + + +@dataclass +class KubernetesNodeInfo: + id: str + + name: str + + +@dataclass +class KubernetesPoolInfo: + id: str + + name: str + + +@dataclass +class SecretManagerSecretInfo: + path: str + + +@dataclass +class SecretManagerSecretVersionInfo: + revision: int + + +@dataclass +class EventPrincipal: + id: str + + +@dataclass +class Resource: + id: str + + type_: ResourceType + + created_at: Optional[datetime] + + updated_at: Optional[datetime] + + deleted_at: Optional[datetime] + + name: Optional[str] + + secm_secret_info: Optional[SecretManagerSecretInfo] + + secm_secret_version_info: Optional[SecretManagerSecretVersionInfo] + + kube_cluster_info: Optional[KubernetesClusterInfo] + + kube_pool_info: Optional[KubernetesPoolInfo] + + kube_node_info: Optional[KubernetesNodeInfo] + + +@dataclass +class Event: + id: str + + locality: str + + organization_id: str + + source_ip: str + + product_name: str + + recorded_at: Optional[datetime] + + principal: Optional[EventPrincipal] + + project_id: Optional[str] + + user_agent: Optional[str] + + service_name: str + + method_name: str + + request_id: str + + status_code: int + + resource: Optional[Resource] + + request_body: Optional[Dict[str, Any]] + + +@dataclass +class Product: + title: str + + name: str + + +@dataclass +class ListEventsRequest: + region: Optional[Region] + """ + Region to target. If none is passed will use default region from the config. + """ + + project_id: Optional[str] + """ + (Optional) ID of the Project containing the Audit Trail events. + """ + + organization_id: Optional[str] + """ + ID of the Organization containing the Audit Trail events. + """ + + resource_type: Optional[ResourceType] + """ + (Optional) Returns a paginated list of Scaleway resources' features. + """ + + method_name: Optional[str] + """ + (Optional) Name of the method or the API call performed. + """ + + status: Optional[int] + """ + (Optional) HTTP status code of the request. Returns either `200` if the request was successful or `403` if the permission was denied. + """ + + recorded_after: Optional[datetime] + """ + (Optional) The `recorded_after` parameter defines the earliest timestamp from which Audit Trail events are retrieved. Returns `one hour ago` by default. + """ + + recorded_before: Optional[datetime] + """ + (Optional) The `recorded_before` parameter defines the latest timestamp up to which Audit Trail events are retrieved. Returns `now` by default. + """ + + order_by: Optional[ListEventsRequestOrderBy] + + page_size: Optional[int] + + page_token: Optional[str] + + product_name: Optional[str] + """ + (Optional) Name of the Scaleway resource in a hyphenated format. + """ + + +@dataclass +class ListEventsResponse: + events: List[Event] + """ + Single page of events matching the requested criteria. + """ + + next_page_token: Optional[str] + """ + Page token to use in following calls to keep listing. + """ + + +@dataclass +class ListProductsRequest: + region: Optional[Region] + """ + Region to target. If none is passed will use default region from the config. + """ + + +@dataclass +class ListProductsResponse: + products: List[Product] + + total_count: int