create_alert async ¶

async def create_alert(
    self,
    space_id: "capo_cloudwatchomni.types.space_id.SpaceId",
    profile_id: "capo_cloudwatchomni.types.profile_id.ProfileId",
    name: str,
    rule: "capo_cloudwatchomni.types.rule.Rule",
    *,
    config_overrides: Optional[AsyncCloudWatchOmniClientConfig] = None,
    description: Optional[str] = None,
    notifications_enabled: Optional[bool] = None,
    tags: Optional["capo_cloudwatchomni.types.tag_map.TagMap"] = None,
    notification_rules: Optional[
        "capo_cloudwatchomni.types.notification_rule_list.NotificationRuleList"
    ] = None,
    client_token: Optional[str] = None,
) -> "capo_cloudwatchomni.types.create_alert_output.CreateAlertOutput":
    r"""Creates a new alert within a space. Use GetAlert and ListAlerts to retrieve alerts, UpdateAlert to modify one, and DeleteAlert to remove it.

    Args:
        space_id: The unique ID of the space to create the alert in.
        profile_id: The ID of the access profile the alert uses to evaluate its query and execute notifications. The caller supplies it: there is no managed alert profile, and the service does not pick one on the caller's behalf.
        name: Alert name, for display. Max 256 (the AlarmName budget). Not the alert's identity: the backend mints a separate uuid as the {@link AlertId}, so the name need not be unique within a space and addressing an alert never depends on it. UpdateAlert accepts a new name to rename the alert.
        description: An optional description of the alert.
        rule: The rule that defines how the alert is evaluated.
        notifications_enabled: Whether actions (notifications) are enabled for this alert. Defaults to true when omitted.
        tags: The tags to associate with the alert.
        notification_rules: The notification rules that determine when and where notifications are sent.
        client_token: Idempotency token for safe retries. Retrying with the same token within the idempotency window returns the original alert instead of creating a duplicate.

    Raises:
        capo_cloudwatchomni.errors.access_denied_exception.AccessDeniedException: The caller is not authorized to perform this action.
        capo_cloudwatchomni.errors.conflict_exception.ConflictException: The operation could not be completed because of a conflict with the current state of the resource.
        capo_cloudwatchomni.errors.internal_server_exception.InternalServerException: An unexpected error occurred while processing the request.
        capo_cloudwatchomni.errors.resource_not_found_exception.ResourceNotFoundException: The specified resource does not exist.
        capo_cloudwatchomni.errors.service_quota_exceeded_exception.ServiceQuotaExceededException: A service quota was exceeded.
        capo_cloudwatchomni.errors.throttling_exception.ThrottlingException: The request was throttled due to exceeding the allowed request rate.
        capo_cloudwatchomni.errors.validation_exception.ValidationException: A parameter is specified incorrectly.
        capo_cloudwatchomni.errors.UnknownServiceError: The service returned an error code this client does not model.

    Examples:
        Create an alert on a field value
        The following example creates an alert whose threshold is compared against a named field of each result row, so every service the query groups by is tracked as its own contributor. FIELD_VALUE requires thresholdField. Payloads are shown as JSON; on the wire they are CBOR-encoded.

        >>> await client.create_alert(space_id='a1b2c3d4-5e6f-4a3b-8c9d-0e1f2a3b4c5d', profile_id='analyst-readonly', name='service-error-count-elevated', description='Alerts when a service logs more errors than its accepted rate.', rule={'telemetryRule': {'query': {'language': 'SQL', 'expression': 'SELECT resource[\'attributes\'][\'service.name\'] AS service, COUNT(*) AS error_count FROM "logs.default" WHERE severityText = \'ERROR\' GROUP BY service'}, 'condition': {'thresholdMode': 'FIELD_VALUE', 'thresholdField': 'error_count', 'comparator': 'GT', 'warningThreshold': 50.0, 'criticalThreshold': 200.0}, 'evaluation': {'intervalSeconds': 300, 'pendingDurationSeconds': 600, 'recoveryDurationSeconds': 300}, 'noData': {'treatAs': 'NODATA'}}}, notifications_enabled=True, notification_rules=[{'trigger': {'stateValues': ['CRITICAL']}, 'target': {'type': 'slack', 'arn': 'arn:aws:cloudwatch:us-east-1:123456789012:integration/a1b2c3d4-5e6f-4a3b-8c9d-0e1f2a3b4c5d', 'metadata': {'channel': 'oncall-alerts'}}}], tags={'Team': 'observability'}, client_token='3f2a9c1e-7b04-4d8a-9e15-6c2b8d0f4a73')
        Create an alert on the number of matching rows
        The following example creates an alert whose threshold is compared against how many rows the query returns, rather than a value within them. COUNT_OF_RESULTS takes no thresholdField. Notifications are created disabled, so the alert evaluates and records state without sending anything, and an empty result set is treated as OK rather than as missing data. Payloads are shown as JSON; on the wire they are CBOR-encoded.

        >>> await client.create_alert(space_id='a1b2c3d4-5e6f-4a3b-8c9d-0e1f2a3b4c5d', profile_id='analyst-readonly', name='service-checkout-5xx-responses', description='Counts checkout responses that returned a server error.', rule={'telemetryRule': {'query': {'language': 'SQL', 'expression': 'SELECT * FROM "logs.default" WHERE resource[\'attributes\'][\'service.name\'] = \'checkout\' AND attributes[\'http.response.status_code\'] >= 500'}, 'condition': {'thresholdMode': 'COUNT_OF_RESULTS', 'comparator': 'GT', 'warningThreshold': 10.0, 'criticalThreshold': 50.0}, 'evaluation': {'intervalSeconds': 60, 'pendingDurationSeconds': 120}, 'noData': {'treatAs': 'OK'}}}, notifications_enabled=False)
        Create an alert from a PromQL query
        The following example creates an alert from a PromQL expression instead of SQL. A PromQL rule compares against the series value, which is carried as the `value` field, so the condition is FIELD_VALUE with thresholdField set to `value`. Notifications go to an Amazon SNS topic, whose ARN is the topic itself rather than an integration. Payloads are shown as JSON; on the wire they are CBOR-encoded.

        >>> await client.create_alert(space_id='a1b2c3d4-5e6f-4a3b-8c9d-0e1f2a3b4c5d', profile_id='analyst-readonly', name='checkout-error-rate-promql', description='Alerts on the checkout server error rate over a five-minute window.', rule={'telemetryRule': {'query': {'language': 'PROMQL', 'expression': 'sum by (service_name) (rate(http_server_errors_total{service_name="checkout"}[5m]))'}, 'condition': {'thresholdMode': 'FIELD_VALUE', 'thresholdField': 'value', 'comparator': 'GT', 'warningThreshold': 0.05, 'criticalThreshold': 0.1}, 'evaluation': {'intervalSeconds': 300, 'pendingDurationSeconds': 300}, 'noData': {'treatAs': 'NODATA'}}}, notification_rules=[{'trigger': {'stateValues': ['WARNING', 'CRITICAL']}, 'target': {'type': 'sns', 'arn': 'arn:aws:sns:us-east-1:123456789012:checkout-oncall'}}])
    """

    async def _handler(
        req: "AsyncOperationRequest[capo_cloudwatchomni.types.create_alert_input.CreateAlertInput]",
    ) -> AsyncOperationResponse[
        "capo_cloudwatchomni.types.create_alert_output.CreateAlertOutput"
    ]:
        import capo_cloudwatchomni._operations.cloud_watch_omni_frontend.create_alert

        (
            output,
            http_response,
        ) = await capo_cloudwatchomni._operations.cloud_watch_omni_frontend.create_alert.async_create_alert(
            req.options, req.input
        )
        return AsyncOperationResponse(output=output, response=http_response)

    interceptors_, options_ = self.operation_options(
        config_overrides, OperationLogging("CreateAlert")
    )
    input_: capo_cloudwatchomni.types.create_alert_input.CreateAlertInput = {
        "space_id": space_id,
        "profile_id": profile_id,
        "name": name,
        "rule": rule,
    }
    if description is not None:
        input_["description"] = description
    if notifications_enabled is not None:
        input_["notifications_enabled"] = notifications_enabled
    if tags is not None:
        input_["tags"] = tags
    if notification_rules is not None:
        input_["notification_rules"] = notification_rules
    if client_token is None:
        client_token = str(uuid.uuid4())
    input_["client_token"] = client_token

    response = await aexecute_pipeline(
        AsyncOperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    await response.response.aclose()
    return response.output