diff --git a/scaleway-async/scaleway_async/audit_trail/v1alpha1/__init__.py b/scaleway-async/scaleway_async/audit_trail/v1alpha1/__init__.py index 1a6969b6c..eed739485 100644 --- a/scaleway-async/scaleway_async/audit_trail/v1alpha1/__init__.py +++ b/scaleway-async/scaleway_async/audit_trail/v1alpha1/__init__.py @@ -7,6 +7,9 @@ from .types import AuthenticationEventMethod from .types import AuthenticationEventOrigin from .types import AuthenticationEventResult +from .types import CustomAlertRuleSeverity +from .types import CustomAlertRuleStatus +from .content import CUSTOM_ALERT_RULE_TRANSIENT_STATUSES from .types import ExportJobStatusCode from .types import ListAuthenticationEventsRequestOrderBy from .types import ListCombinedEventsRequestOrderBy @@ -71,15 +74,22 @@ from .types import ExportJobStatus from .types import ProductService from .types import AlertRule +from .types import CustomAlertRule from .types import ListCombinedEventsResponseCombinedEvent from .types import ExportJob from .types import Product +from .types import CreateCustomAlertRuleRequest from .types import CreateExportJobRequest +from .types import DeleteCustomAlertRuleRequest from .types import DeleteExportJobRequest from .types import DisableAlertRulesRequest from .types import DisableAlertRulesResponse +from .types import DisableCustomAlertRulesRequest +from .types import DisableCustomAlertRulesResponse from .types import EnableAlertRulesRequest from .types import EnableAlertRulesResponse +from .types import EnableCustomAlertRulesRequest +from .types import EnableCustomAlertRulesResponse from .types import EventsOverview from .types import GetLastEventsOverviewRequest from .types import ListAlertRulesRequest @@ -88,6 +98,8 @@ from .types import ListAuthenticationEventsResponse from .types import ListCombinedEventsRequest from .types import ListCombinedEventsResponse +from .types import ListCustomAlertRulesRequest +from .types import ListCustomAlertRulesResponse from .types import ListEventsRequest from .types import ListEventsResponse from .types import ListExportJobsRequest @@ -98,6 +110,9 @@ from .types import ListSystemEventsResponse from .types import SetEnabledAlertRulesRequest from .types import SetEnabledAlertRulesResponse +from .types import SetEnabledCustomAlertRulesRequest +from .types import SetEnabledCustomAlertRulesResponse +from .types import UpdateCustomAlertRuleRequest from .api import AuditTrailV1Alpha1API __all__ = [ @@ -108,6 +123,9 @@ "AuthenticationEventMethod", "AuthenticationEventOrigin", "AuthenticationEventResult", + "CustomAlertRuleSeverity", + "CustomAlertRuleStatus", + "CUSTOM_ALERT_RULE_TRANSIENT_STATUSES", "ExportJobStatusCode", "ListAuthenticationEventsRequestOrderBy", "ListCombinedEventsRequestOrderBy", @@ -172,15 +190,22 @@ "ExportJobStatus", "ProductService", "AlertRule", + "CustomAlertRule", "ListCombinedEventsResponseCombinedEvent", "ExportJob", "Product", + "CreateCustomAlertRuleRequest", "CreateExportJobRequest", + "DeleteCustomAlertRuleRequest", "DeleteExportJobRequest", "DisableAlertRulesRequest", "DisableAlertRulesResponse", + "DisableCustomAlertRulesRequest", + "DisableCustomAlertRulesResponse", "EnableAlertRulesRequest", "EnableAlertRulesResponse", + "EnableCustomAlertRulesRequest", + "EnableCustomAlertRulesResponse", "EventsOverview", "GetLastEventsOverviewRequest", "ListAlertRulesRequest", @@ -189,6 +214,8 @@ "ListAuthenticationEventsResponse", "ListCombinedEventsRequest", "ListCombinedEventsResponse", + "ListCustomAlertRulesRequest", + "ListCustomAlertRulesResponse", "ListEventsRequest", "ListEventsResponse", "ListExportJobsRequest", @@ -199,5 +226,8 @@ "ListSystemEventsResponse", "SetEnabledAlertRulesRequest", "SetEnabledAlertRulesResponse", + "SetEnabledCustomAlertRulesRequest", + "SetEnabledCustomAlertRulesResponse", + "UpdateCustomAlertRuleRequest", "AuditTrailV1Alpha1API", ] diff --git a/scaleway-async/scaleway_async/audit_trail/v1alpha1/api.py b/scaleway-async/scaleway_async/audit_trail/v1alpha1/api.py index fa077243e..0086a4ec5 100644 --- a/scaleway-async/scaleway_async/audit_trail/v1alpha1/api.py +++ b/scaleway-async/scaleway_async/audit_trail/v1alpha1/api.py @@ -14,6 +14,8 @@ ) from .types import ( AlertRuleStatus, + CustomAlertRuleSeverity, + CustomAlertRuleStatus, ListAuthenticationEventsRequestOrderBy, ListCombinedEventsRequestOrderBy, ListEventsRequestOrderBy, @@ -21,41 +23,61 @@ ListSystemEventsRequestOrderBy, ResourceType, AlertRule, + CreateCustomAlertRuleRequest, CreateExportJobRequest, + CustomAlertRule, DisableAlertRulesRequest, DisableAlertRulesResponse, + DisableCustomAlertRulesRequest, + DisableCustomAlertRulesResponse, EnableAlertRulesRequest, EnableAlertRulesResponse, + EnableCustomAlertRulesRequest, + EnableCustomAlertRulesResponse, EventsOverview, ExportJob, ExportJobS3, ListAlertRulesResponse, ListAuthenticationEventsResponse, ListCombinedEventsResponse, + ListCustomAlertRulesResponse, ListEventsResponse, ListExportJobsResponse, ListProductsResponse, ListSystemEventsResponse, SetEnabledAlertRulesRequest, SetEnabledAlertRulesResponse, + SetEnabledCustomAlertRulesRequest, + SetEnabledCustomAlertRulesResponse, + UpdateCustomAlertRuleRequest, ) from .marshalling import ( + unmarshal_CustomAlertRule, unmarshal_ExportJob, unmarshal_DisableAlertRulesResponse, + unmarshal_DisableCustomAlertRulesResponse, unmarshal_EnableAlertRulesResponse, + unmarshal_EnableCustomAlertRulesResponse, unmarshal_EventsOverview, unmarshal_ListAlertRulesResponse, unmarshal_ListAuthenticationEventsResponse, unmarshal_ListCombinedEventsResponse, + unmarshal_ListCustomAlertRulesResponse, unmarshal_ListEventsResponse, unmarshal_ListExportJobsResponse, unmarshal_ListProductsResponse, unmarshal_ListSystemEventsResponse, unmarshal_SetEnabledAlertRulesResponse, + unmarshal_SetEnabledCustomAlertRulesResponse, + marshal_CreateCustomAlertRuleRequest, marshal_CreateExportJobRequest, marshal_DisableAlertRulesRequest, + marshal_DisableCustomAlertRulesRequest, marshal_EnableAlertRulesRequest, + marshal_EnableCustomAlertRulesRequest, marshal_SetEnabledAlertRulesRequest, + marshal_SetEnabledCustomAlertRulesRequest, + marshal_UpdateCustomAlertRuleRequest, ) @@ -617,6 +639,86 @@ async def list_alert_rules_all( }, ) + async def list_custom_alert_rules( + self, + *, + region: Optional[ScwRegion] = None, + organization_id: Optional[str] = None, + status: Optional[CustomAlertRuleStatus] = None, + page: Optional[int] = None, + page_size: Optional[int] = None, + ) -> ListCustomAlertRulesResponse: + """ + List custom alert rules for a specified organization and their current status (enabled or disabled). + :param region: Region to target. If none is passed will use default region from the config. + :param organization_id: ID of the Organization to target. + :param status: (Optional) Status of the custom alert rule. + :param page: + :param page_size: + :return: :class:`ListCustomAlertRulesResponse ` + + Usage: + :: + + result = await api.list_custom_alert_rules() + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "GET", + f"/audit-trail/v1alpha1/regions/{param_region}/custom-alert-rules", + params={ + "organization_id": organization_id + or self.client.default_organization_id, + "page": page, + "page_size": page_size or self.client.default_page_size, + "status": status, + }, + ) + + self._throw_on_error(res) + return unmarshal_ListCustomAlertRulesResponse(res.json()) + + async def list_custom_alert_rules_all( + self, + *, + region: Optional[ScwRegion] = None, + organization_id: Optional[str] = None, + status: Optional[CustomAlertRuleStatus] = None, + page: Optional[int] = None, + page_size: Optional[int] = None, + ) -> list[CustomAlertRule]: + """ + List custom alert rules for a specified organization and their current status (enabled or disabled). + :param region: Region to target. If none is passed will use default region from the config. + :param organization_id: ID of the Organization to target. + :param status: (Optional) Status of the custom alert rule. + :param page: + :param page_size: + :return: :class:`list[CustomAlertRule] ` + + Usage: + :: + + result = await api.list_custom_alert_rules_all() + """ + + return await fetch_all_pages_async( + type=ListCustomAlertRulesResponse, + key="custom_alert_rules", + fetcher=self.list_custom_alert_rules, + args={ + "region": region, + "organization_id": organization_id, + "status": status, + "page": page, + "page_size": page_size, + }, + ) + async def enable_alert_rules( self, *, @@ -658,6 +760,47 @@ async def enable_alert_rules( self._throw_on_error(res) return unmarshal_EnableAlertRulesResponse(res.json()) + async def enable_custom_alert_rules( + self, + *, + region: Optional[ScwRegion] = None, + organization_id: Optional[str] = None, + custom_alert_rule_ids: Optional[list[str]] = None, + ) -> EnableCustomAlertRulesResponse: + """ + Enable custom alert rules. + Enable custom alert rules for a specified organization. Enabled custom rules will trigger alerts when matching events occur. + :param region: Region to target. If none is passed will use default region from the config. + :param organization_id: ID of the Organization to target. + :param custom_alert_rule_ids: List of IDs of the custom rules to enable. + :return: :class:`EnableCustomAlertRulesResponse ` + + Usage: + :: + + result = await api.enable_custom_alert_rules() + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "POST", + f"/audit-trail/v1alpha1/regions/{param_region}/enable-custom-alert-rules", + body=marshal_EnableCustomAlertRulesRequest( + EnableCustomAlertRulesRequest( + region=region, + organization_id=organization_id, + custom_alert_rule_ids=custom_alert_rule_ids, + ), + self.client, + ), + ) + + self._throw_on_error(res) + return unmarshal_EnableCustomAlertRulesResponse(res.json()) + async def disable_alert_rules( self, *, @@ -699,6 +842,47 @@ async def disable_alert_rules( self._throw_on_error(res) return unmarshal_DisableAlertRulesResponse(res.json()) + async def disable_custom_alert_rules( + self, + *, + region: Optional[ScwRegion] = None, + organization_id: Optional[str] = None, + custom_alert_rule_ids: Optional[list[str]] = None, + ) -> DisableCustomAlertRulesResponse: + """ + Disable custom alert rules. + Disable custom alert rules for a specified organization. Disabled rules will no longer trigger alerts when matching events occur. + :param region: Region to target. If none is passed will use default region from the config. + :param organization_id: ID of the Organization to target. + :param custom_alert_rule_ids: List of IDs of the custom rules to disable. + :return: :class:`DisableCustomAlertRulesResponse ` + + Usage: + :: + + result = await api.disable_custom_alert_rules() + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "POST", + f"/audit-trail/v1alpha1/regions/{param_region}/disable-custom-alert-rules", + body=marshal_DisableCustomAlertRulesRequest( + DisableCustomAlertRulesRequest( + region=region, + organization_id=organization_id, + custom_alert_rule_ids=custom_alert_rule_ids, + ), + self.client, + ), + ) + + self._throw_on_error(res) + return unmarshal_DisableCustomAlertRulesResponse(res.json()) + async def set_enabled_alert_rules( self, *, @@ -739,3 +923,187 @@ async def set_enabled_alert_rules( self._throw_on_error(res) return unmarshal_SetEnabledAlertRulesResponse(res.json()) + + async def set_enabled_custom_alert_rules( + self, + *, + region: Optional[ScwRegion] = None, + organization_id: Optional[str] = None, + enabled_custom_alert_rule_ids: Optional[list[str]] = None, + ) -> SetEnabledCustomAlertRulesResponse: + """ + Set the custom alert rules to enabled. + Set the custom alert rules to enabled by replacing the set of enabled custom alert rules for a specified organization. The provided list defines the complete set of custom rules that should be enabled. Any previously enabled custom rule not included in the request will be disabled. + :param region: Region to target. If none is passed will use default region from the config. + :param organization_id: ID of the Organization to target. + :param enabled_custom_alert_rule_ids: List of IDs of the custom rules that must be enabled after the update. + :return: :class:`SetEnabledCustomAlertRulesResponse ` + + Usage: + :: + + result = await api.set_enabled_custom_alert_rules() + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "PUT", + f"/audit-trail/v1alpha1/regions/{param_region}/custom-alert-rules", + body=marshal_SetEnabledCustomAlertRulesRequest( + SetEnabledCustomAlertRulesRequest( + region=region, + organization_id=organization_id, + enabled_custom_alert_rule_ids=enabled_custom_alert_rule_ids, + ), + self.client, + ), + ) + + self._throw_on_error(res) + return unmarshal_SetEnabledCustomAlertRulesResponse(res.json()) + + async def create_custom_alert_rule( + self, + *, + name: str, + query: str, + occurrences: int, + region: Optional[ScwRegion] = None, + organization_id: Optional[str] = None, + description: Optional[str] = None, + evaluation_window: Optional[str] = None, + severity: Optional[CustomAlertRuleSeverity] = None, + ) -> CustomAlertRule: + """ + Create a custom alert rule. + Create a custom alert rule in a given region specified by the `region` parameter. + :param name: Name of the custom alert rule. + :param query: The Common Expression Language (CEL) string defining the logic for the alert rule. + :param occurrences: The minimum number of matched occurrences required within the evaluation window to trigger the alert. + :param region: Region to target. If none is passed will use default region from the config. + :param organization_id: ID of the Organization to target. + :param description: (Optional) Description of the custom alert rule. + :param evaluation_window: The duration of time over which to evaluate the rule (how far back to look for matching events). + :param severity: (Optional) The severity level assigned to the custom alert rule. By default, the severity will be set to info. + :return: :class:`CustomAlertRule ` + + Usage: + :: + + result = await api.create_custom_alert_rule( + name="example", + query="example", + occurrences=1, + ) + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "POST", + f"/audit-trail/v1alpha1/regions/{param_region}/custom-alert-rules", + body=marshal_CreateCustomAlertRuleRequest( + CreateCustomAlertRuleRequest( + name=name, + query=query, + occurrences=occurrences, + region=region, + organization_id=organization_id, + description=description, + evaluation_window=evaluation_window, + severity=severity, + ), + self.client, + ), + ) + + self._throw_on_error(res) + return unmarshal_CustomAlertRule(res.json()) + + async def update_custom_alert_rule( + self, + *, + custom_alert_rule_id: str, + region: Optional[ScwRegion] = None, + name: Optional[str] = None, + description: Optional[str] = None, + ) -> CustomAlertRule: + """ + Update a custom alert rule. + Modify a custom alert rule's metadata including name and description, specified by the `alert_rule_id` and `region` parameters. + :param custom_alert_rule_id: ID of the custom alert rule to update. + :param region: Region to target. If none is passed will use default region from the config. + :param name: (Optional) New name for the custom alert rule. + :param description: (Optional) New description for the custom alert rule. + :return: :class:`CustomAlertRule ` + + Usage: + :: + + result = await api.update_custom_alert_rule( + custom_alert_rule_id="example", + ) + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + param_custom_alert_rule_id = validate_path_param( + "custom_alert_rule_id", custom_alert_rule_id + ) + + res = self._request( + "PATCH", + f"/audit-trail/v1alpha1/regions/{param_region}/custom-alert-rules/{param_custom_alert_rule_id}", + body=marshal_UpdateCustomAlertRuleRequest( + UpdateCustomAlertRuleRequest( + custom_alert_rule_id=custom_alert_rule_id, + region=region, + name=name, + description=description, + ), + self.client, + ), + ) + + self._throw_on_error(res) + return unmarshal_CustomAlertRule(res.json()) + + async def delete_custom_alert_rule( + self, + *, + custom_alert_rule_id: str, + region: Optional[ScwRegion] = None, + ) -> None: + """ + Delete a custom alert rule. + Permanently delete a custom alert rule specified by the `region` and `alert_rule_id` parameters. This action is irreversible. + :param custom_alert_rule_id: ID of the custom alert rule to delete. + :param region: Region to target. If none is passed will use default region from the config. + + Usage: + :: + + result = await api.delete_custom_alert_rule( + custom_alert_rule_id="example", + ) + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + param_custom_alert_rule_id = validate_path_param( + "custom_alert_rule_id", custom_alert_rule_id + ) + + res = self._request( + "DELETE", + f"/audit-trail/v1alpha1/regions/{param_region}/custom-alert-rules/{param_custom_alert_rule_id}", + ) + + self._throw_on_error(res) diff --git a/scaleway-async/scaleway_async/audit_trail/v1alpha1/content.py b/scaleway-async/scaleway_async/audit_trail/v1alpha1/content.py index 3fb5eefa7..a34ff007c 100644 --- a/scaleway-async/scaleway_async/audit_trail/v1alpha1/content.py +++ b/scaleway-async/scaleway_async/audit_trail/v1alpha1/content.py @@ -3,6 +3,7 @@ from .types import ( AlertRuleStatus, + CustomAlertRuleStatus, ) ALERT_RULE_TRANSIENT_STATUSES: list[AlertRuleStatus] = [ @@ -12,3 +13,10 @@ """ Lists transient statutes of the enum :class:`AlertRuleStatus `. """ +CUSTOM_ALERT_RULE_TRANSIENT_STATUSES: list[CustomAlertRuleStatus] = [ + CustomAlertRuleStatus.ENABLING, + CustomAlertRuleStatus.DISABLING, +] +""" +Lists transient statutes of the enum :class:`CustomAlertRuleStatus `. +""" diff --git a/scaleway-async/scaleway_async/audit_trail/v1alpha1/marshalling.py b/scaleway-async/scaleway_async/audit_trail/v1alpha1/marshalling.py index 5c34b58c4..b769543a3 100644 --- a/scaleway-async/scaleway_async/audit_trail/v1alpha1/marshalling.py +++ b/scaleway-async/scaleway_async/audit_trail/v1alpha1/marshalling.py @@ -16,13 +16,18 @@ AuthenticationEventMethod, AuthenticationEventOrigin, AuthenticationEventResult, + CustomAlertRuleSeverity, + CustomAlertRuleStatus, SystemEventKind, + CustomAlertRule, ExportJobS3, ExportJobStatus, ExportJob, AlertRule, DisableAlertRulesResponse, + DisableCustomAlertRulesResponse, EnableAlertRulesResponse, + EnableCustomAlertRulesResponse, AccountContractSignatureInfoAccountContractInfo, AccountContractSignatureInfo, AccountOrganizationInfo, @@ -80,6 +85,7 @@ SystemEvent, ListCombinedEventsResponseCombinedEvent, ListCombinedEventsResponse, + ListCustomAlertRulesResponse, ListEventsResponse, ListExportJobsResponse, ProductService, @@ -87,16 +93,93 @@ ListProductsResponse, ListSystemEventsResponse, SetEnabledAlertRulesResponse, + SetEnabledCustomAlertRulesResponse, + CreateCustomAlertRuleRequest, CreateExportJobRequest, DisableAlertRulesRequest, + DisableCustomAlertRulesRequest, EnableAlertRulesRequest, + EnableCustomAlertRulesRequest, SetEnabledAlertRulesRequest, + SetEnabledCustomAlertRulesRequest, + UpdateCustomAlertRuleRequest, ) from ...std.types import ( CountryCode as StdCountryCode, ) +def unmarshal_CustomAlertRule(data: Any) -> CustomAlertRule: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'CustomAlertRule' 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("name", None) + if field is not None: + args["name"] = field + else: + args["name"] = None + + field = data.get("status", None) + if field is not None: + args["status"] = field + else: + args["status"] = CustomAlertRuleStatus.UNKNOWN_STATUS + + field = data.get("query", None) + if field is not None: + args["query"] = field + else: + args["query"] = None + + field = data.get("occurrences", None) + if field is not None: + args["occurrences"] = field + else: + args["occurrences"] = 0 + + field = data.get("severity", None) + if field is not None: + args["severity"] = field + else: + args["severity"] = CustomAlertRuleSeverity.UNKNOWN_SEVERITY + + field = data.get("description", None) + if field is not None: + args["description"] = field + else: + args["description"] = None + + field = data.get("evaluation_window", None) + if field is not None: + args["evaluation_window"] = field + else: + args["evaluation_window"] = None + + 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 + + return CustomAlertRule(**args) + + def unmarshal_ExportJobS3(data: Any) -> ExportJobS3: if not isinstance(data, dict): raise TypeError( @@ -270,6 +353,27 @@ def unmarshal_DisableAlertRulesResponse(data: Any) -> DisableAlertRulesResponse: return DisableAlertRulesResponse(**args) +def unmarshal_DisableCustomAlertRulesResponse( + data: Any, +) -> DisableCustomAlertRulesResponse: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'DisableCustomAlertRulesResponse' failed as data isn't a dictionary." + ) + + args: dict[str, Any] = {} + + field = data.get("custom_alert_rules", None) + if field is not None: + args["custom_alert_rules"] = ( + [unmarshal_CustomAlertRule(v) for v in field] if field is not None else None + ) + else: + args["custom_alert_rules"] = [] + + return DisableCustomAlertRulesResponse(**args) + + def unmarshal_EnableAlertRulesResponse(data: Any) -> EnableAlertRulesResponse: if not isinstance(data, dict): raise TypeError( @@ -289,6 +393,27 @@ def unmarshal_EnableAlertRulesResponse(data: Any) -> EnableAlertRulesResponse: return EnableAlertRulesResponse(**args) +def unmarshal_EnableCustomAlertRulesResponse( + data: Any, +) -> EnableCustomAlertRulesResponse: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'EnableCustomAlertRulesResponse' failed as data isn't a dictionary." + ) + + args: dict[str, Any] = {} + + field = data.get("custom_alert_rules", None) + if field is not None: + args["custom_alert_rules"] = ( + [unmarshal_CustomAlertRule(v) for v in field] if field is not None else None + ) + else: + args["custom_alert_rules"] = [] + + return EnableCustomAlertRulesResponse(**args) + + def unmarshal_AccountContractSignatureInfoAccountContractInfo( data: Any, ) -> AccountContractSignatureInfoAccountContractInfo: @@ -2044,6 +2169,31 @@ def unmarshal_ListCombinedEventsResponse(data: Any) -> ListCombinedEventsRespons return ListCombinedEventsResponse(**args) +def unmarshal_ListCustomAlertRulesResponse(data: Any) -> ListCustomAlertRulesResponse: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'ListCustomAlertRulesResponse' failed as data isn't a dictionary." + ) + + args: dict[str, Any] = {} + + field = data.get("custom_alert_rules", None) + if field is not None: + args["custom_alert_rules"] = ( + [unmarshal_CustomAlertRule(v) for v in field] if field is not None else None + ) + else: + args["custom_alert_rules"] = [] + + field = data.get("total_count", None) + if field is not None: + args["total_count"] = field + else: + args["total_count"] = 0 + + return ListCustomAlertRulesResponse(**args) + + def unmarshal_ListEventsResponse(data: Any) -> ListEventsResponse: if not isinstance(data, dict): raise TypeError( @@ -2217,6 +2367,59 @@ def unmarshal_SetEnabledAlertRulesResponse(data: Any) -> SetEnabledAlertRulesRes return SetEnabledAlertRulesResponse(**args) +def unmarshal_SetEnabledCustomAlertRulesResponse( + data: Any, +) -> SetEnabledCustomAlertRulesResponse: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'SetEnabledCustomAlertRulesResponse' failed as data isn't a dictionary." + ) + + args: dict[str, Any] = {} + + field = data.get("custom_alert_rules", None) + if field is not None: + args["custom_alert_rules"] = ( + [unmarshal_CustomAlertRule(v) for v in field] if field is not None else None + ) + else: + args["custom_alert_rules"] = [] + + return SetEnabledCustomAlertRulesResponse(**args) + + +def marshal_CreateCustomAlertRuleRequest( + request: CreateCustomAlertRuleRequest, + defaults: ProfileDefaults, +) -> dict[str, Any]: + output: dict[str, Any] = {} + + if request.name is not None: + output["name"] = request.name + + if request.query is not None: + output["query"] = request.query + + if request.occurrences is not None: + output["occurrences"] = request.occurrences + + if request.organization_id is not None: + output["organization_id"] = request.organization_id + else: + output["organization_id"] = defaults.default_organization_id + + if request.description is not None: + output["description"] = request.description + + if request.evaluation_window is not None: + output["evaluation_window"] = request.evaluation_window + + if request.severity is not None: + output["severity"] = request.severity + + return output + + def marshal_ExportJobS3( request: ExportJobS3, defaults: ProfileDefaults, @@ -2286,6 +2489,23 @@ def marshal_DisableAlertRulesRequest( return output +def marshal_DisableCustomAlertRulesRequest( + request: DisableCustomAlertRulesRequest, + defaults: ProfileDefaults, +) -> dict[str, Any]: + output: dict[str, Any] = {} + + if request.organization_id is not None: + output["organization_id"] = request.organization_id + else: + output["organization_id"] = defaults.default_organization_id + + if request.custom_alert_rule_ids is not None: + output["custom_alert_rule_ids"] = request.custom_alert_rule_ids + + return output + + def marshal_EnableAlertRulesRequest( request: EnableAlertRulesRequest, defaults: ProfileDefaults, @@ -2303,6 +2523,23 @@ def marshal_EnableAlertRulesRequest( return output +def marshal_EnableCustomAlertRulesRequest( + request: EnableCustomAlertRulesRequest, + defaults: ProfileDefaults, +) -> dict[str, Any]: + output: dict[str, Any] = {} + + if request.organization_id is not None: + output["organization_id"] = request.organization_id + else: + output["organization_id"] = defaults.default_organization_id + + if request.custom_alert_rule_ids is not None: + output["custom_alert_rule_ids"] = request.custom_alert_rule_ids + + return output + + def marshal_SetEnabledAlertRulesRequest( request: SetEnabledAlertRulesRequest, defaults: ProfileDefaults, @@ -2318,3 +2555,35 @@ def marshal_SetEnabledAlertRulesRequest( output["enabled_alert_rule_ids"] = request.enabled_alert_rule_ids return output + + +def marshal_SetEnabledCustomAlertRulesRequest( + request: SetEnabledCustomAlertRulesRequest, + defaults: ProfileDefaults, +) -> dict[str, Any]: + output: dict[str, Any] = {} + + if request.organization_id is not None: + output["organization_id"] = request.organization_id + else: + output["organization_id"] = defaults.default_organization_id + + if request.enabled_custom_alert_rule_ids is not None: + output["enabled_custom_alert_rule_ids"] = request.enabled_custom_alert_rule_ids + + return output + + +def marshal_UpdateCustomAlertRuleRequest( + request: UpdateCustomAlertRuleRequest, + defaults: ProfileDefaults, +) -> dict[str, Any]: + output: dict[str, Any] = {} + + if request.name is not None: + output["name"] = request.name + + if request.description is not None: + output["description"] = request.description + + return output diff --git a/scaleway-async/scaleway_async/audit_trail/v1alpha1/types.py b/scaleway-async/scaleway_async/audit_trail/v1alpha1/types.py index 00fec0259..964aed01a 100644 --- a/scaleway-async/scaleway_async/audit_trail/v1alpha1/types.py +++ b/scaleway-async/scaleway_async/audit_trail/v1alpha1/types.py @@ -76,6 +76,28 @@ def __str__(self) -> str: return str(self.value) +class CustomAlertRuleSeverity(str, Enum, metaclass=StrEnumMeta): + UNKNOWN_SEVERITY = "unknown_severity" + INFO = "info" + ERROR = "error" + WARNING = "warning" + CRITICAL = "critical" + + def __str__(self) -> str: + return str(self.value) + + +class CustomAlertRuleStatus(str, Enum, metaclass=StrEnumMeta): + UNKNOWN_STATUS = "unknown_status" + ENABLED = "enabled" + DISABLED = "disabled" + ENABLING = "enabling" + DISABLING = "disabling" + + def __str__(self) -> str: + return str(self.value) + + class ExportJobStatusCode(str, Enum, metaclass=StrEnumMeta): UNKNOWN_CODE = "unknown_code" SUCCESS = "success" @@ -872,6 +894,59 @@ class AlertRule: """ +@dataclass +class CustomAlertRule: + id: str + """ + ID of the alert rule. + """ + + name: str + """ + Name of the alert rule. + """ + + status: CustomAlertRuleStatus + """ + Current status of the alert rule. + """ + + query: str + """ + The Common Expression Language (CEL) string defining the logic for the alert rule. + """ + + occurrences: int + """ + The minimum number of matched occurrences required within the evaluation window to trigger the alert. + """ + + severity: CustomAlertRuleSeverity + """ + The severity level assigned to the custom alert rule. + """ + + description: Optional[str] = None + """ + (Optional) Description of the alert rule. + """ + + evaluation_window: Optional[str] = None + """ + The duration of time over which to evaluate the rule (how far back to look for matching events). + """ + + created_at: Optional[datetime] = None + """ + Custom alert rule creation date. + """ + + updated_at: Optional[datetime] = None + """ + Custom alert rule last modification date. + """ + + @dataclass class ListCombinedEventsResponseCombinedEvent: api: Optional[Event] = None @@ -939,6 +1014,51 @@ class Product: """ +@dataclass +class CreateCustomAlertRuleRequest: + name: str + """ + Name of the custom alert rule. + """ + + query: str + """ + The Common Expression Language (CEL) string defining the logic for the alert rule. + """ + + occurrences: int + """ + The minimum number of matched occurrences required within the evaluation window to trigger the alert. + """ + + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + organization_id: Optional[str] = None + """ + ID of the Organization to target. + """ + + description: Optional[str] = None + """ + (Optional) Description of the custom alert rule. + """ + + evaluation_window: Optional[str] = None + """ + The duration of time over which to evaluate the rule (how far back to look for matching events). + """ + + severity: Optional[CustomAlertRuleSeverity] = ( + CustomAlertRuleSeverity.UNKNOWN_SEVERITY + ) + """ + (Optional) The severity level assigned to the custom alert rule. By default, the severity will be set to info. + """ + + @dataclass class CreateExportJobRequest: name: str @@ -964,6 +1084,19 @@ class CreateExportJobRequest: s3: Optional[ExportJobS3] = None +@dataclass +class DeleteCustomAlertRuleRequest: + custom_alert_rule_id: str + """ + ID of the custom alert rule to delete. + """ + + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + @dataclass class DeleteExportJobRequest: export_job_id: str @@ -999,7 +1132,33 @@ class DisableAlertRulesRequest: class DisableAlertRulesResponse: alert_rules: list[AlertRule] """ - List of the rules that were disabled. + List of the preconfigured rules that were disabled. + """ + + +@dataclass +class DisableCustomAlertRulesRequest: + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + organization_id: Optional[str] = None + """ + ID of the Organization to target. + """ + + custom_alert_rule_ids: Optional[list[str]] = field(default_factory=list) + """ + List of IDs of the custom rules to disable. + """ + + +@dataclass +class DisableCustomAlertRulesResponse: + custom_alert_rules: list[CustomAlertRule] + """ + List of the custom rules that were disabled. """ @@ -1025,7 +1184,33 @@ class EnableAlertRulesRequest: class EnableAlertRulesResponse: alert_rules: list[AlertRule] """ - List of the rules that were enabled. + List of the preconfigured rules that were enabled. + """ + + +@dataclass +class EnableCustomAlertRulesRequest: + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + organization_id: Optional[str] = None + """ + ID of the Organization to target. + """ + + custom_alert_rule_ids: Optional[list[str]] = field(default_factory=list) + """ + List of IDs of the custom rules to enable. + """ + + +@dataclass +class EnableCustomAlertRulesResponse: + custom_alert_rules: list[CustomAlertRule] + """ + List of the custom rules that were enabled. """ @@ -1070,7 +1255,7 @@ class ListAlertRulesRequest: class ListAlertRulesResponse: alert_rules: list[AlertRule] """ - Single page of alert rules matching the requested criteria. + Single page of preconfigured alert rules matching the requested criteria. """ total_count: int @@ -1123,6 +1308,40 @@ class ListCombinedEventsResponse: next_page_token: Optional[str] = None +@dataclass +class ListCustomAlertRulesRequest: + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + organization_id: Optional[str] = None + """ + ID of the Organization to target. + """ + + status: Optional[CustomAlertRuleStatus] = CustomAlertRuleStatus.UNKNOWN_STATUS + """ + (Optional) Status of the custom alert rule. + """ + + page: Optional[int] = 0 + page_size: Optional[int] = 0 + + +@dataclass +class ListCustomAlertRulesResponse: + custom_alert_rules: list[CustomAlertRule] + """ + Single page of custom alert rules matching the requested criteria. + """ + + total_count: int + """ + Total count of custom alert rules matching the requested criteria. + """ + + @dataclass class ListEventsRequest: region: Optional[ScwRegion] = None @@ -1341,5 +1560,54 @@ class SetEnabledAlertRulesRequest: class SetEnabledAlertRulesResponse: alert_rules: list[AlertRule] """ - List of the rules that were enabled. + List of the preconfigured rules that were enabled. + """ + + +@dataclass +class SetEnabledCustomAlertRulesRequest: + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + organization_id: Optional[str] = None + """ + ID of the Organization to target. + """ + + enabled_custom_alert_rule_ids: Optional[list[str]] = field(default_factory=list) + """ + List of IDs of the custom rules that must be enabled after the update. + """ + + +@dataclass +class SetEnabledCustomAlertRulesResponse: + custom_alert_rules: list[CustomAlertRule] + """ + List of the custom rules that were enabled. + """ + + +@dataclass +class UpdateCustomAlertRuleRequest: + custom_alert_rule_id: str + """ + ID of the custom alert rule to update. + """ + + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + name: Optional[str] = None + """ + (Optional) New name for the custom alert rule. + """ + + description: Optional[str] = None + """ + (Optional) New description for the custom alert rule. """ diff --git a/scaleway/scaleway/audit_trail/v1alpha1/__init__.py b/scaleway/scaleway/audit_trail/v1alpha1/__init__.py index 1a6969b6c..eed739485 100644 --- a/scaleway/scaleway/audit_trail/v1alpha1/__init__.py +++ b/scaleway/scaleway/audit_trail/v1alpha1/__init__.py @@ -7,6 +7,9 @@ from .types import AuthenticationEventMethod from .types import AuthenticationEventOrigin from .types import AuthenticationEventResult +from .types import CustomAlertRuleSeverity +from .types import CustomAlertRuleStatus +from .content import CUSTOM_ALERT_RULE_TRANSIENT_STATUSES from .types import ExportJobStatusCode from .types import ListAuthenticationEventsRequestOrderBy from .types import ListCombinedEventsRequestOrderBy @@ -71,15 +74,22 @@ from .types import ExportJobStatus from .types import ProductService from .types import AlertRule +from .types import CustomAlertRule from .types import ListCombinedEventsResponseCombinedEvent from .types import ExportJob from .types import Product +from .types import CreateCustomAlertRuleRequest from .types import CreateExportJobRequest +from .types import DeleteCustomAlertRuleRequest from .types import DeleteExportJobRequest from .types import DisableAlertRulesRequest from .types import DisableAlertRulesResponse +from .types import DisableCustomAlertRulesRequest +from .types import DisableCustomAlertRulesResponse from .types import EnableAlertRulesRequest from .types import EnableAlertRulesResponse +from .types import EnableCustomAlertRulesRequest +from .types import EnableCustomAlertRulesResponse from .types import EventsOverview from .types import GetLastEventsOverviewRequest from .types import ListAlertRulesRequest @@ -88,6 +98,8 @@ from .types import ListAuthenticationEventsResponse from .types import ListCombinedEventsRequest from .types import ListCombinedEventsResponse +from .types import ListCustomAlertRulesRequest +from .types import ListCustomAlertRulesResponse from .types import ListEventsRequest from .types import ListEventsResponse from .types import ListExportJobsRequest @@ -98,6 +110,9 @@ from .types import ListSystemEventsResponse from .types import SetEnabledAlertRulesRequest from .types import SetEnabledAlertRulesResponse +from .types import SetEnabledCustomAlertRulesRequest +from .types import SetEnabledCustomAlertRulesResponse +from .types import UpdateCustomAlertRuleRequest from .api import AuditTrailV1Alpha1API __all__ = [ @@ -108,6 +123,9 @@ "AuthenticationEventMethod", "AuthenticationEventOrigin", "AuthenticationEventResult", + "CustomAlertRuleSeverity", + "CustomAlertRuleStatus", + "CUSTOM_ALERT_RULE_TRANSIENT_STATUSES", "ExportJobStatusCode", "ListAuthenticationEventsRequestOrderBy", "ListCombinedEventsRequestOrderBy", @@ -172,15 +190,22 @@ "ExportJobStatus", "ProductService", "AlertRule", + "CustomAlertRule", "ListCombinedEventsResponseCombinedEvent", "ExportJob", "Product", + "CreateCustomAlertRuleRequest", "CreateExportJobRequest", + "DeleteCustomAlertRuleRequest", "DeleteExportJobRequest", "DisableAlertRulesRequest", "DisableAlertRulesResponse", + "DisableCustomAlertRulesRequest", + "DisableCustomAlertRulesResponse", "EnableAlertRulesRequest", "EnableAlertRulesResponse", + "EnableCustomAlertRulesRequest", + "EnableCustomAlertRulesResponse", "EventsOverview", "GetLastEventsOverviewRequest", "ListAlertRulesRequest", @@ -189,6 +214,8 @@ "ListAuthenticationEventsResponse", "ListCombinedEventsRequest", "ListCombinedEventsResponse", + "ListCustomAlertRulesRequest", + "ListCustomAlertRulesResponse", "ListEventsRequest", "ListEventsResponse", "ListExportJobsRequest", @@ -199,5 +226,8 @@ "ListSystemEventsResponse", "SetEnabledAlertRulesRequest", "SetEnabledAlertRulesResponse", + "SetEnabledCustomAlertRulesRequest", + "SetEnabledCustomAlertRulesResponse", + "UpdateCustomAlertRuleRequest", "AuditTrailV1Alpha1API", ] diff --git a/scaleway/scaleway/audit_trail/v1alpha1/api.py b/scaleway/scaleway/audit_trail/v1alpha1/api.py index bf56f66ac..dda5855dd 100644 --- a/scaleway/scaleway/audit_trail/v1alpha1/api.py +++ b/scaleway/scaleway/audit_trail/v1alpha1/api.py @@ -14,6 +14,8 @@ ) from .types import ( AlertRuleStatus, + CustomAlertRuleSeverity, + CustomAlertRuleStatus, ListAuthenticationEventsRequestOrderBy, ListCombinedEventsRequestOrderBy, ListEventsRequestOrderBy, @@ -21,41 +23,61 @@ ListSystemEventsRequestOrderBy, ResourceType, AlertRule, + CreateCustomAlertRuleRequest, CreateExportJobRequest, + CustomAlertRule, DisableAlertRulesRequest, DisableAlertRulesResponse, + DisableCustomAlertRulesRequest, + DisableCustomAlertRulesResponse, EnableAlertRulesRequest, EnableAlertRulesResponse, + EnableCustomAlertRulesRequest, + EnableCustomAlertRulesResponse, EventsOverview, ExportJob, ExportJobS3, ListAlertRulesResponse, ListAuthenticationEventsResponse, ListCombinedEventsResponse, + ListCustomAlertRulesResponse, ListEventsResponse, ListExportJobsResponse, ListProductsResponse, ListSystemEventsResponse, SetEnabledAlertRulesRequest, SetEnabledAlertRulesResponse, + SetEnabledCustomAlertRulesRequest, + SetEnabledCustomAlertRulesResponse, + UpdateCustomAlertRuleRequest, ) from .marshalling import ( + unmarshal_CustomAlertRule, unmarshal_ExportJob, unmarshal_DisableAlertRulesResponse, + unmarshal_DisableCustomAlertRulesResponse, unmarshal_EnableAlertRulesResponse, + unmarshal_EnableCustomAlertRulesResponse, unmarshal_EventsOverview, unmarshal_ListAlertRulesResponse, unmarshal_ListAuthenticationEventsResponse, unmarshal_ListCombinedEventsResponse, + unmarshal_ListCustomAlertRulesResponse, unmarshal_ListEventsResponse, unmarshal_ListExportJobsResponse, unmarshal_ListProductsResponse, unmarshal_ListSystemEventsResponse, unmarshal_SetEnabledAlertRulesResponse, + unmarshal_SetEnabledCustomAlertRulesResponse, + marshal_CreateCustomAlertRuleRequest, marshal_CreateExportJobRequest, marshal_DisableAlertRulesRequest, + marshal_DisableCustomAlertRulesRequest, marshal_EnableAlertRulesRequest, + marshal_EnableCustomAlertRulesRequest, marshal_SetEnabledAlertRulesRequest, + marshal_SetEnabledCustomAlertRulesRequest, + marshal_UpdateCustomAlertRuleRequest, ) @@ -617,6 +639,86 @@ def list_alert_rules_all( }, ) + def list_custom_alert_rules( + self, + *, + region: Optional[ScwRegion] = None, + organization_id: Optional[str] = None, + status: Optional[CustomAlertRuleStatus] = None, + page: Optional[int] = None, + page_size: Optional[int] = None, + ) -> ListCustomAlertRulesResponse: + """ + List custom alert rules for a specified organization and their current status (enabled or disabled). + :param region: Region to target. If none is passed will use default region from the config. + :param organization_id: ID of the Organization to target. + :param status: (Optional) Status of the custom alert rule. + :param page: + :param page_size: + :return: :class:`ListCustomAlertRulesResponse ` + + Usage: + :: + + result = api.list_custom_alert_rules() + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "GET", + f"/audit-trail/v1alpha1/regions/{param_region}/custom-alert-rules", + params={ + "organization_id": organization_id + or self.client.default_organization_id, + "page": page, + "page_size": page_size or self.client.default_page_size, + "status": status, + }, + ) + + self._throw_on_error(res) + return unmarshal_ListCustomAlertRulesResponse(res.json()) + + def list_custom_alert_rules_all( + self, + *, + region: Optional[ScwRegion] = None, + organization_id: Optional[str] = None, + status: Optional[CustomAlertRuleStatus] = None, + page: Optional[int] = None, + page_size: Optional[int] = None, + ) -> list[CustomAlertRule]: + """ + List custom alert rules for a specified organization and their current status (enabled or disabled). + :param region: Region to target. If none is passed will use default region from the config. + :param organization_id: ID of the Organization to target. + :param status: (Optional) Status of the custom alert rule. + :param page: + :param page_size: + :return: :class:`list[CustomAlertRule] ` + + Usage: + :: + + result = api.list_custom_alert_rules_all() + """ + + return fetch_all_pages( + type=ListCustomAlertRulesResponse, + key="custom_alert_rules", + fetcher=self.list_custom_alert_rules, + args={ + "region": region, + "organization_id": organization_id, + "status": status, + "page": page, + "page_size": page_size, + }, + ) + def enable_alert_rules( self, *, @@ -658,6 +760,47 @@ def enable_alert_rules( self._throw_on_error(res) return unmarshal_EnableAlertRulesResponse(res.json()) + def enable_custom_alert_rules( + self, + *, + region: Optional[ScwRegion] = None, + organization_id: Optional[str] = None, + custom_alert_rule_ids: Optional[list[str]] = None, + ) -> EnableCustomAlertRulesResponse: + """ + Enable custom alert rules. + Enable custom alert rules for a specified organization. Enabled custom rules will trigger alerts when matching events occur. + :param region: Region to target. If none is passed will use default region from the config. + :param organization_id: ID of the Organization to target. + :param custom_alert_rule_ids: List of IDs of the custom rules to enable. + :return: :class:`EnableCustomAlertRulesResponse ` + + Usage: + :: + + result = api.enable_custom_alert_rules() + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "POST", + f"/audit-trail/v1alpha1/regions/{param_region}/enable-custom-alert-rules", + body=marshal_EnableCustomAlertRulesRequest( + EnableCustomAlertRulesRequest( + region=region, + organization_id=organization_id, + custom_alert_rule_ids=custom_alert_rule_ids, + ), + self.client, + ), + ) + + self._throw_on_error(res) + return unmarshal_EnableCustomAlertRulesResponse(res.json()) + def disable_alert_rules( self, *, @@ -699,6 +842,47 @@ def disable_alert_rules( self._throw_on_error(res) return unmarshal_DisableAlertRulesResponse(res.json()) + def disable_custom_alert_rules( + self, + *, + region: Optional[ScwRegion] = None, + organization_id: Optional[str] = None, + custom_alert_rule_ids: Optional[list[str]] = None, + ) -> DisableCustomAlertRulesResponse: + """ + Disable custom alert rules. + Disable custom alert rules for a specified organization. Disabled rules will no longer trigger alerts when matching events occur. + :param region: Region to target. If none is passed will use default region from the config. + :param organization_id: ID of the Organization to target. + :param custom_alert_rule_ids: List of IDs of the custom rules to disable. + :return: :class:`DisableCustomAlertRulesResponse ` + + Usage: + :: + + result = api.disable_custom_alert_rules() + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "POST", + f"/audit-trail/v1alpha1/regions/{param_region}/disable-custom-alert-rules", + body=marshal_DisableCustomAlertRulesRequest( + DisableCustomAlertRulesRequest( + region=region, + organization_id=organization_id, + custom_alert_rule_ids=custom_alert_rule_ids, + ), + self.client, + ), + ) + + self._throw_on_error(res) + return unmarshal_DisableCustomAlertRulesResponse(res.json()) + def set_enabled_alert_rules( self, *, @@ -739,3 +923,187 @@ def set_enabled_alert_rules( self._throw_on_error(res) return unmarshal_SetEnabledAlertRulesResponse(res.json()) + + def set_enabled_custom_alert_rules( + self, + *, + region: Optional[ScwRegion] = None, + organization_id: Optional[str] = None, + enabled_custom_alert_rule_ids: Optional[list[str]] = None, + ) -> SetEnabledCustomAlertRulesResponse: + """ + Set the custom alert rules to enabled. + Set the custom alert rules to enabled by replacing the set of enabled custom alert rules for a specified organization. The provided list defines the complete set of custom rules that should be enabled. Any previously enabled custom rule not included in the request will be disabled. + :param region: Region to target. If none is passed will use default region from the config. + :param organization_id: ID of the Organization to target. + :param enabled_custom_alert_rule_ids: List of IDs of the custom rules that must be enabled after the update. + :return: :class:`SetEnabledCustomAlertRulesResponse ` + + Usage: + :: + + result = api.set_enabled_custom_alert_rules() + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "PUT", + f"/audit-trail/v1alpha1/regions/{param_region}/custom-alert-rules", + body=marshal_SetEnabledCustomAlertRulesRequest( + SetEnabledCustomAlertRulesRequest( + region=region, + organization_id=organization_id, + enabled_custom_alert_rule_ids=enabled_custom_alert_rule_ids, + ), + self.client, + ), + ) + + self._throw_on_error(res) + return unmarshal_SetEnabledCustomAlertRulesResponse(res.json()) + + def create_custom_alert_rule( + self, + *, + name: str, + query: str, + occurrences: int, + region: Optional[ScwRegion] = None, + organization_id: Optional[str] = None, + description: Optional[str] = None, + evaluation_window: Optional[str] = None, + severity: Optional[CustomAlertRuleSeverity] = None, + ) -> CustomAlertRule: + """ + Create a custom alert rule. + Create a custom alert rule in a given region specified by the `region` parameter. + :param name: Name of the custom alert rule. + :param query: The Common Expression Language (CEL) string defining the logic for the alert rule. + :param occurrences: The minimum number of matched occurrences required within the evaluation window to trigger the alert. + :param region: Region to target. If none is passed will use default region from the config. + :param organization_id: ID of the Organization to target. + :param description: (Optional) Description of the custom alert rule. + :param evaluation_window: The duration of time over which to evaluate the rule (how far back to look for matching events). + :param severity: (Optional) The severity level assigned to the custom alert rule. By default, the severity will be set to info. + :return: :class:`CustomAlertRule ` + + Usage: + :: + + result = api.create_custom_alert_rule( + name="example", + query="example", + occurrences=1, + ) + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + + res = self._request( + "POST", + f"/audit-trail/v1alpha1/regions/{param_region}/custom-alert-rules", + body=marshal_CreateCustomAlertRuleRequest( + CreateCustomAlertRuleRequest( + name=name, + query=query, + occurrences=occurrences, + region=region, + organization_id=organization_id, + description=description, + evaluation_window=evaluation_window, + severity=severity, + ), + self.client, + ), + ) + + self._throw_on_error(res) + return unmarshal_CustomAlertRule(res.json()) + + def update_custom_alert_rule( + self, + *, + custom_alert_rule_id: str, + region: Optional[ScwRegion] = None, + name: Optional[str] = None, + description: Optional[str] = None, + ) -> CustomAlertRule: + """ + Update a custom alert rule. + Modify a custom alert rule's metadata including name and description, specified by the `alert_rule_id` and `region` parameters. + :param custom_alert_rule_id: ID of the custom alert rule to update. + :param region: Region to target. If none is passed will use default region from the config. + :param name: (Optional) New name for the custom alert rule. + :param description: (Optional) New description for the custom alert rule. + :return: :class:`CustomAlertRule ` + + Usage: + :: + + result = api.update_custom_alert_rule( + custom_alert_rule_id="example", + ) + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + param_custom_alert_rule_id = validate_path_param( + "custom_alert_rule_id", custom_alert_rule_id + ) + + res = self._request( + "PATCH", + f"/audit-trail/v1alpha1/regions/{param_region}/custom-alert-rules/{param_custom_alert_rule_id}", + body=marshal_UpdateCustomAlertRuleRequest( + UpdateCustomAlertRuleRequest( + custom_alert_rule_id=custom_alert_rule_id, + region=region, + name=name, + description=description, + ), + self.client, + ), + ) + + self._throw_on_error(res) + return unmarshal_CustomAlertRule(res.json()) + + def delete_custom_alert_rule( + self, + *, + custom_alert_rule_id: str, + region: Optional[ScwRegion] = None, + ) -> None: + """ + Delete a custom alert rule. + Permanently delete a custom alert rule specified by the `region` and `alert_rule_id` parameters. This action is irreversible. + :param custom_alert_rule_id: ID of the custom alert rule to delete. + :param region: Region to target. If none is passed will use default region from the config. + + Usage: + :: + + result = api.delete_custom_alert_rule( + custom_alert_rule_id="example", + ) + """ + + param_region = validate_path_param( + "region", region or self.client.default_region + ) + param_custom_alert_rule_id = validate_path_param( + "custom_alert_rule_id", custom_alert_rule_id + ) + + res = self._request( + "DELETE", + f"/audit-trail/v1alpha1/regions/{param_region}/custom-alert-rules/{param_custom_alert_rule_id}", + ) + + self._throw_on_error(res) diff --git a/scaleway/scaleway/audit_trail/v1alpha1/content.py b/scaleway/scaleway/audit_trail/v1alpha1/content.py index 3fb5eefa7..a34ff007c 100644 --- a/scaleway/scaleway/audit_trail/v1alpha1/content.py +++ b/scaleway/scaleway/audit_trail/v1alpha1/content.py @@ -3,6 +3,7 @@ from .types import ( AlertRuleStatus, + CustomAlertRuleStatus, ) ALERT_RULE_TRANSIENT_STATUSES: list[AlertRuleStatus] = [ @@ -12,3 +13,10 @@ """ Lists transient statutes of the enum :class:`AlertRuleStatus `. """ +CUSTOM_ALERT_RULE_TRANSIENT_STATUSES: list[CustomAlertRuleStatus] = [ + CustomAlertRuleStatus.ENABLING, + CustomAlertRuleStatus.DISABLING, +] +""" +Lists transient statutes of the enum :class:`CustomAlertRuleStatus `. +""" diff --git a/scaleway/scaleway/audit_trail/v1alpha1/marshalling.py b/scaleway/scaleway/audit_trail/v1alpha1/marshalling.py index 5c34b58c4..b769543a3 100644 --- a/scaleway/scaleway/audit_trail/v1alpha1/marshalling.py +++ b/scaleway/scaleway/audit_trail/v1alpha1/marshalling.py @@ -16,13 +16,18 @@ AuthenticationEventMethod, AuthenticationEventOrigin, AuthenticationEventResult, + CustomAlertRuleSeverity, + CustomAlertRuleStatus, SystemEventKind, + CustomAlertRule, ExportJobS3, ExportJobStatus, ExportJob, AlertRule, DisableAlertRulesResponse, + DisableCustomAlertRulesResponse, EnableAlertRulesResponse, + EnableCustomAlertRulesResponse, AccountContractSignatureInfoAccountContractInfo, AccountContractSignatureInfo, AccountOrganizationInfo, @@ -80,6 +85,7 @@ SystemEvent, ListCombinedEventsResponseCombinedEvent, ListCombinedEventsResponse, + ListCustomAlertRulesResponse, ListEventsResponse, ListExportJobsResponse, ProductService, @@ -87,16 +93,93 @@ ListProductsResponse, ListSystemEventsResponse, SetEnabledAlertRulesResponse, + SetEnabledCustomAlertRulesResponse, + CreateCustomAlertRuleRequest, CreateExportJobRequest, DisableAlertRulesRequest, + DisableCustomAlertRulesRequest, EnableAlertRulesRequest, + EnableCustomAlertRulesRequest, SetEnabledAlertRulesRequest, + SetEnabledCustomAlertRulesRequest, + UpdateCustomAlertRuleRequest, ) from ...std.types import ( CountryCode as StdCountryCode, ) +def unmarshal_CustomAlertRule(data: Any) -> CustomAlertRule: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'CustomAlertRule' 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("name", None) + if field is not None: + args["name"] = field + else: + args["name"] = None + + field = data.get("status", None) + if field is not None: + args["status"] = field + else: + args["status"] = CustomAlertRuleStatus.UNKNOWN_STATUS + + field = data.get("query", None) + if field is not None: + args["query"] = field + else: + args["query"] = None + + field = data.get("occurrences", None) + if field is not None: + args["occurrences"] = field + else: + args["occurrences"] = 0 + + field = data.get("severity", None) + if field is not None: + args["severity"] = field + else: + args["severity"] = CustomAlertRuleSeverity.UNKNOWN_SEVERITY + + field = data.get("description", None) + if field is not None: + args["description"] = field + else: + args["description"] = None + + field = data.get("evaluation_window", None) + if field is not None: + args["evaluation_window"] = field + else: + args["evaluation_window"] = None + + 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 + + return CustomAlertRule(**args) + + def unmarshal_ExportJobS3(data: Any) -> ExportJobS3: if not isinstance(data, dict): raise TypeError( @@ -270,6 +353,27 @@ def unmarshal_DisableAlertRulesResponse(data: Any) -> DisableAlertRulesResponse: return DisableAlertRulesResponse(**args) +def unmarshal_DisableCustomAlertRulesResponse( + data: Any, +) -> DisableCustomAlertRulesResponse: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'DisableCustomAlertRulesResponse' failed as data isn't a dictionary." + ) + + args: dict[str, Any] = {} + + field = data.get("custom_alert_rules", None) + if field is not None: + args["custom_alert_rules"] = ( + [unmarshal_CustomAlertRule(v) for v in field] if field is not None else None + ) + else: + args["custom_alert_rules"] = [] + + return DisableCustomAlertRulesResponse(**args) + + def unmarshal_EnableAlertRulesResponse(data: Any) -> EnableAlertRulesResponse: if not isinstance(data, dict): raise TypeError( @@ -289,6 +393,27 @@ def unmarshal_EnableAlertRulesResponse(data: Any) -> EnableAlertRulesResponse: return EnableAlertRulesResponse(**args) +def unmarshal_EnableCustomAlertRulesResponse( + data: Any, +) -> EnableCustomAlertRulesResponse: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'EnableCustomAlertRulesResponse' failed as data isn't a dictionary." + ) + + args: dict[str, Any] = {} + + field = data.get("custom_alert_rules", None) + if field is not None: + args["custom_alert_rules"] = ( + [unmarshal_CustomAlertRule(v) for v in field] if field is not None else None + ) + else: + args["custom_alert_rules"] = [] + + return EnableCustomAlertRulesResponse(**args) + + def unmarshal_AccountContractSignatureInfoAccountContractInfo( data: Any, ) -> AccountContractSignatureInfoAccountContractInfo: @@ -2044,6 +2169,31 @@ def unmarshal_ListCombinedEventsResponse(data: Any) -> ListCombinedEventsRespons return ListCombinedEventsResponse(**args) +def unmarshal_ListCustomAlertRulesResponse(data: Any) -> ListCustomAlertRulesResponse: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'ListCustomAlertRulesResponse' failed as data isn't a dictionary." + ) + + args: dict[str, Any] = {} + + field = data.get("custom_alert_rules", None) + if field is not None: + args["custom_alert_rules"] = ( + [unmarshal_CustomAlertRule(v) for v in field] if field is not None else None + ) + else: + args["custom_alert_rules"] = [] + + field = data.get("total_count", None) + if field is not None: + args["total_count"] = field + else: + args["total_count"] = 0 + + return ListCustomAlertRulesResponse(**args) + + def unmarshal_ListEventsResponse(data: Any) -> ListEventsResponse: if not isinstance(data, dict): raise TypeError( @@ -2217,6 +2367,59 @@ def unmarshal_SetEnabledAlertRulesResponse(data: Any) -> SetEnabledAlertRulesRes return SetEnabledAlertRulesResponse(**args) +def unmarshal_SetEnabledCustomAlertRulesResponse( + data: Any, +) -> SetEnabledCustomAlertRulesResponse: + if not isinstance(data, dict): + raise TypeError( + "Unmarshalling the type 'SetEnabledCustomAlertRulesResponse' failed as data isn't a dictionary." + ) + + args: dict[str, Any] = {} + + field = data.get("custom_alert_rules", None) + if field is not None: + args["custom_alert_rules"] = ( + [unmarshal_CustomAlertRule(v) for v in field] if field is not None else None + ) + else: + args["custom_alert_rules"] = [] + + return SetEnabledCustomAlertRulesResponse(**args) + + +def marshal_CreateCustomAlertRuleRequest( + request: CreateCustomAlertRuleRequest, + defaults: ProfileDefaults, +) -> dict[str, Any]: + output: dict[str, Any] = {} + + if request.name is not None: + output["name"] = request.name + + if request.query is not None: + output["query"] = request.query + + if request.occurrences is not None: + output["occurrences"] = request.occurrences + + if request.organization_id is not None: + output["organization_id"] = request.organization_id + else: + output["organization_id"] = defaults.default_organization_id + + if request.description is not None: + output["description"] = request.description + + if request.evaluation_window is not None: + output["evaluation_window"] = request.evaluation_window + + if request.severity is not None: + output["severity"] = request.severity + + return output + + def marshal_ExportJobS3( request: ExportJobS3, defaults: ProfileDefaults, @@ -2286,6 +2489,23 @@ def marshal_DisableAlertRulesRequest( return output +def marshal_DisableCustomAlertRulesRequest( + request: DisableCustomAlertRulesRequest, + defaults: ProfileDefaults, +) -> dict[str, Any]: + output: dict[str, Any] = {} + + if request.organization_id is not None: + output["organization_id"] = request.organization_id + else: + output["organization_id"] = defaults.default_organization_id + + if request.custom_alert_rule_ids is not None: + output["custom_alert_rule_ids"] = request.custom_alert_rule_ids + + return output + + def marshal_EnableAlertRulesRequest( request: EnableAlertRulesRequest, defaults: ProfileDefaults, @@ -2303,6 +2523,23 @@ def marshal_EnableAlertRulesRequest( return output +def marshal_EnableCustomAlertRulesRequest( + request: EnableCustomAlertRulesRequest, + defaults: ProfileDefaults, +) -> dict[str, Any]: + output: dict[str, Any] = {} + + if request.organization_id is not None: + output["organization_id"] = request.organization_id + else: + output["organization_id"] = defaults.default_organization_id + + if request.custom_alert_rule_ids is not None: + output["custom_alert_rule_ids"] = request.custom_alert_rule_ids + + return output + + def marshal_SetEnabledAlertRulesRequest( request: SetEnabledAlertRulesRequest, defaults: ProfileDefaults, @@ -2318,3 +2555,35 @@ def marshal_SetEnabledAlertRulesRequest( output["enabled_alert_rule_ids"] = request.enabled_alert_rule_ids return output + + +def marshal_SetEnabledCustomAlertRulesRequest( + request: SetEnabledCustomAlertRulesRequest, + defaults: ProfileDefaults, +) -> dict[str, Any]: + output: dict[str, Any] = {} + + if request.organization_id is not None: + output["organization_id"] = request.organization_id + else: + output["organization_id"] = defaults.default_organization_id + + if request.enabled_custom_alert_rule_ids is not None: + output["enabled_custom_alert_rule_ids"] = request.enabled_custom_alert_rule_ids + + return output + + +def marshal_UpdateCustomAlertRuleRequest( + request: UpdateCustomAlertRuleRequest, + defaults: ProfileDefaults, +) -> dict[str, Any]: + output: dict[str, Any] = {} + + if request.name is not None: + output["name"] = request.name + + if request.description is not None: + output["description"] = request.description + + return output diff --git a/scaleway/scaleway/audit_trail/v1alpha1/types.py b/scaleway/scaleway/audit_trail/v1alpha1/types.py index 00fec0259..964aed01a 100644 --- a/scaleway/scaleway/audit_trail/v1alpha1/types.py +++ b/scaleway/scaleway/audit_trail/v1alpha1/types.py @@ -76,6 +76,28 @@ def __str__(self) -> str: return str(self.value) +class CustomAlertRuleSeverity(str, Enum, metaclass=StrEnumMeta): + UNKNOWN_SEVERITY = "unknown_severity" + INFO = "info" + ERROR = "error" + WARNING = "warning" + CRITICAL = "critical" + + def __str__(self) -> str: + return str(self.value) + + +class CustomAlertRuleStatus(str, Enum, metaclass=StrEnumMeta): + UNKNOWN_STATUS = "unknown_status" + ENABLED = "enabled" + DISABLED = "disabled" + ENABLING = "enabling" + DISABLING = "disabling" + + def __str__(self) -> str: + return str(self.value) + + class ExportJobStatusCode(str, Enum, metaclass=StrEnumMeta): UNKNOWN_CODE = "unknown_code" SUCCESS = "success" @@ -872,6 +894,59 @@ class AlertRule: """ +@dataclass +class CustomAlertRule: + id: str + """ + ID of the alert rule. + """ + + name: str + """ + Name of the alert rule. + """ + + status: CustomAlertRuleStatus + """ + Current status of the alert rule. + """ + + query: str + """ + The Common Expression Language (CEL) string defining the logic for the alert rule. + """ + + occurrences: int + """ + The minimum number of matched occurrences required within the evaluation window to trigger the alert. + """ + + severity: CustomAlertRuleSeverity + """ + The severity level assigned to the custom alert rule. + """ + + description: Optional[str] = None + """ + (Optional) Description of the alert rule. + """ + + evaluation_window: Optional[str] = None + """ + The duration of time over which to evaluate the rule (how far back to look for matching events). + """ + + created_at: Optional[datetime] = None + """ + Custom alert rule creation date. + """ + + updated_at: Optional[datetime] = None + """ + Custom alert rule last modification date. + """ + + @dataclass class ListCombinedEventsResponseCombinedEvent: api: Optional[Event] = None @@ -939,6 +1014,51 @@ class Product: """ +@dataclass +class CreateCustomAlertRuleRequest: + name: str + """ + Name of the custom alert rule. + """ + + query: str + """ + The Common Expression Language (CEL) string defining the logic for the alert rule. + """ + + occurrences: int + """ + The minimum number of matched occurrences required within the evaluation window to trigger the alert. + """ + + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + organization_id: Optional[str] = None + """ + ID of the Organization to target. + """ + + description: Optional[str] = None + """ + (Optional) Description of the custom alert rule. + """ + + evaluation_window: Optional[str] = None + """ + The duration of time over which to evaluate the rule (how far back to look for matching events). + """ + + severity: Optional[CustomAlertRuleSeverity] = ( + CustomAlertRuleSeverity.UNKNOWN_SEVERITY + ) + """ + (Optional) The severity level assigned to the custom alert rule. By default, the severity will be set to info. + """ + + @dataclass class CreateExportJobRequest: name: str @@ -964,6 +1084,19 @@ class CreateExportJobRequest: s3: Optional[ExportJobS3] = None +@dataclass +class DeleteCustomAlertRuleRequest: + custom_alert_rule_id: str + """ + ID of the custom alert rule to delete. + """ + + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + @dataclass class DeleteExportJobRequest: export_job_id: str @@ -999,7 +1132,33 @@ class DisableAlertRulesRequest: class DisableAlertRulesResponse: alert_rules: list[AlertRule] """ - List of the rules that were disabled. + List of the preconfigured rules that were disabled. + """ + + +@dataclass +class DisableCustomAlertRulesRequest: + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + organization_id: Optional[str] = None + """ + ID of the Organization to target. + """ + + custom_alert_rule_ids: Optional[list[str]] = field(default_factory=list) + """ + List of IDs of the custom rules to disable. + """ + + +@dataclass +class DisableCustomAlertRulesResponse: + custom_alert_rules: list[CustomAlertRule] + """ + List of the custom rules that were disabled. """ @@ -1025,7 +1184,33 @@ class EnableAlertRulesRequest: class EnableAlertRulesResponse: alert_rules: list[AlertRule] """ - List of the rules that were enabled. + List of the preconfigured rules that were enabled. + """ + + +@dataclass +class EnableCustomAlertRulesRequest: + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + organization_id: Optional[str] = None + """ + ID of the Organization to target. + """ + + custom_alert_rule_ids: Optional[list[str]] = field(default_factory=list) + """ + List of IDs of the custom rules to enable. + """ + + +@dataclass +class EnableCustomAlertRulesResponse: + custom_alert_rules: list[CustomAlertRule] + """ + List of the custom rules that were enabled. """ @@ -1070,7 +1255,7 @@ class ListAlertRulesRequest: class ListAlertRulesResponse: alert_rules: list[AlertRule] """ - Single page of alert rules matching the requested criteria. + Single page of preconfigured alert rules matching the requested criteria. """ total_count: int @@ -1123,6 +1308,40 @@ class ListCombinedEventsResponse: next_page_token: Optional[str] = None +@dataclass +class ListCustomAlertRulesRequest: + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + organization_id: Optional[str] = None + """ + ID of the Organization to target. + """ + + status: Optional[CustomAlertRuleStatus] = CustomAlertRuleStatus.UNKNOWN_STATUS + """ + (Optional) Status of the custom alert rule. + """ + + page: Optional[int] = 0 + page_size: Optional[int] = 0 + + +@dataclass +class ListCustomAlertRulesResponse: + custom_alert_rules: list[CustomAlertRule] + """ + Single page of custom alert rules matching the requested criteria. + """ + + total_count: int + """ + Total count of custom alert rules matching the requested criteria. + """ + + @dataclass class ListEventsRequest: region: Optional[ScwRegion] = None @@ -1341,5 +1560,54 @@ class SetEnabledAlertRulesRequest: class SetEnabledAlertRulesResponse: alert_rules: list[AlertRule] """ - List of the rules that were enabled. + List of the preconfigured rules that were enabled. + """ + + +@dataclass +class SetEnabledCustomAlertRulesRequest: + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + organization_id: Optional[str] = None + """ + ID of the Organization to target. + """ + + enabled_custom_alert_rule_ids: Optional[list[str]] = field(default_factory=list) + """ + List of IDs of the custom rules that must be enabled after the update. + """ + + +@dataclass +class SetEnabledCustomAlertRulesResponse: + custom_alert_rules: list[CustomAlertRule] + """ + List of the custom rules that were enabled. + """ + + +@dataclass +class UpdateCustomAlertRuleRequest: + custom_alert_rule_id: str + """ + ID of the custom alert rule to update. + """ + + region: Optional[ScwRegion] = None + """ + Region to target. If none is passed will use default region from the config. + """ + + name: Optional[str] = None + """ + (Optional) New name for the custom alert rule. + """ + + description: Optional[str] = None + """ + (Optional) New description for the custom alert rule. """