Repository: keephq/keep
Pull request: feat: batch alerts enrichment using cel
Closes #4567

## 📑 Description

Implement a new endpoint that allows users to enrich multiple alerts at once by specifying a CEL (Common Expression Language) query instead of listing individual alert fingerprints.

## Usage

```bash
curl --location '127.0.0.1:8080/alerts/batch_enrich?dispose_on_new_alert=true' \
--header 'Content-Type: application/json' \
--header 'X-API-KEY: 5df8d505-dc43-49a6-ba3d-5ccaeb8c7e68' \
--data '{
    "enrichments": {
        "status": "resolved"
    },
    "cel": "name=='\''Pod api-service-production lacks memory'\''"
}'
```

## Additional

https://github.com/user-attachments/assets/a6bd1bd8-7a29-4dc1-83cd-21f055a1c626



Problem statement:
[➕ Feature]: Batch Enrichment by CEL Expression
This is ChatGPT generated description

## Description
Implement a new endpoint that allows users to enrich multiple alerts at once by specifying a CEL (Common Expression Language) query instead of listing individual alert fingerprints.

## Use Cases
- Enrich alerts by attributes like severity or name pattern
- Target alerts with specific labels or properties
- Bulk acknowledge alerts from specific regions or services
- Apply status changes to groups of related alerts

## Examples
```
// Acknowledge all critical severity alerts
{
  "cel": "labels.severity == 'critical'",
  "enrichments": {
    "status": "acknowledged"
  }
}

// Assign all memory-related alerts in a specific region
{
  "cel": "name.contains('Memory') && labels.region == 'us-east-1'",
  "enrichments": {
    "assignee": "infrastructure-team@example.com"
  }
}
```

Hints from issue discussion:

Commit message: fix: revert generating openapi docs

Patch to review:
diff --git a/keep/api/models/alert.py b/keep/api/models/alert.py
index 2203f59cef..5cdfc149ef 100644
--- a/keep/api/models/alert.py
+++ b/keep/api/models/alert.py
@@ -367,7 +367,10 @@ class EnrichAlertRequestBody(BaseModel):
 
 class BatchEnrichAlertRequestBody(BaseModel):
     enrichments: dict[str, str]
-    fingerprints: list[str]
+    fingerprints: Optional[list[str]] = None
+    cel: Optional[str] = None
+    limit: int = 1000
+    offset: int = 0
 
 
 class UnEnrichAlertRequestBody(BaseModel):
diff --git a/keep/api/routes/alerts.py b/keep/api/routes/alerts.py
index 6352191170..d746fef310 100644
--- a/keep/api/routes/alerts.py
+++ b/keep/api/routes/alerts.py
@@ -778,7 +778,7 @@ def enrich_alert_note(
 
 @router.post(
     "/batch_enrich",
-    description="Enrich an alerts",
+    description="Enrich alerts by providing either a list of fingerprints or a CEL expression to select alerts. Examples for CEL: \"name.contains('CPU')\", \"labels.severity == 'critical'\", \"name.contains('Memory') && labels.region == 'us-east-1'\"",
 )
 def batch_enrich_alerts(
     enrich_data: BatchEnrichAlertRequestBody,
@@ -792,9 +792,8 @@ def batch_enrich_alerts(
 ):
     tenant_id = authenticated_entity.tenant_id
     logger.info(
-        "Enriching alerts batch",
+        "Enriching alerts in batch",
         extra={
-            "fingerprints": enrich_data.fingerprints,
             "tenant_id": tenant_id,
         },
     )
@@ -805,6 +804,78 @@ def batch_enrich_alerts(
     ):
         enrich_data.enrichments["status"] = AlertStatus.SUPPRESSED.value
 
+    if not enrich_data.fingerprints and not enrich_data.cel:
+        raise HTTPException(
+            status_code=400, detail="Either fingerprints or cel must be provided"
+        )
+
+    if enrich_data.fingerprints and enrich_data.cel:
+        raise HTTPException(
+            status_code=400, detail="Either fingerprints or cel can be provided at once"
+        )
+
+    # If CEL is provided, use it to find matching alerts
+    if enrich_data.cel:
+        logger.info(
+            "Enriching alerts by CEL query",
+            extra={
+                "cel": enrich_data.cel,
+                "tenant_id": tenant_id,
+            },
+        )
+
+        try:
+            db_alerts, total_count = query_last_alerts(
+                tenant_id=tenant_id,
+                query=QueryDto(
+                    cel=enrich_data.cel,
+                    limit=enrich_data.limit,
+                    offset=enrich_data.offset,
+                ),
+            )
+
+            if not db_alerts:
+                logger.info(
+                    "No alerts found matching the CEL query",
+                    extra={"cel": enrich_data.cel, "tenant_id": tenant_id},
+                )
+                return {
+                    "status": "ok",
+                    "message": "No alerts matched the query",
+                }
+
+            fingerprints = [alert.fingerprint for alert in db_alerts]
+            logger.info(
+                "Found alerts matching CEL query",
+                extra={
+                    "cel": enrich_data.cel,
+                    "tenant_id": tenant_id,
+                    "alert_count": total_count,
+                },
+            )
+        except CelToSqlException as e:
+            logger.exception(
+                f'Error parsing CEL expression "{enrich_data.cel}". {str(e)}'
+            )
+            raise HTTPException(
+                status_code=400,
+                detail=f"Error parsing CEL expression: {enrich_data.cel}",
+            ) from e
+        except Exception as e:
+            logger.exception("Failed to process CEL query", extra={"error": str(e)})
+            return {"status": "failed", "message": str(e)}
+    else:
+        # Use the provided fingerprints
+        fingerprints = enrich_data.fingerprints
+        logger.info(
+            "Enriching alerts batch",
+            extra={
+                "fingerprints": fingerprints,
+                "tenant_id": tenant_id,
+            },
+        )
+
+    # Common enrichment processing
     try:
         enrichment_bl = EnrichmentsBl(tenant_id, db=session)
         (
@@ -819,7 +890,7 @@ def batch_enrich_alerts(
         enrichments = deepcopy(enrich_data.enrichments)
 
         enrichment_bl.batch_enrich(
-            fingerprints=enrich_data.fingerprints,
+            fingerprints=fingerprints,
             enrichments=enrichments,
             action_type=action_type,
             action_callee=authenticated_entity.email,
@@ -828,9 +899,8 @@ def batch_enrich_alerts(
         )
 
         last_alerts = get_last_alerts_by_fingerprints(
-            authenticated_entity.tenant_id, enrich_data.fingerprints, session=session
+            tenant_id, fingerprints, session=session
         )
-
         alert_ids = [last_alert.alert_id for last_alert in last_alerts]
 
         if dispose_on_new_alert:
@@ -843,20 +913,16 @@ def batch_enrich_alerts(
                 )
                 for alert_id in alert_ids
             ]
-
             enrichment_bl.batch_enrich(
                 fingerprints=formatted_alert_ids,
-                enrichments=enrich_data.enrichments,
+                enrichments=enrichments,
                 action_type=action_type,
                 action_callee=authenticated_entity.email,
                 action_description=action_description,
                 audit_enabled=False,
             )
 
-        # get the alert with the new enrichment
-        alerts = get_alerts_by_ids(
-            authenticated_entity.tenant_id, alert_ids, session=session
-        )
+        alerts = get_alerts_by_ids(tenant_id, alert_ids, session=session)
 
         enriched_alerts_dto = convert_db_alerts_to_dto_alerts(alerts, session=session)
         # push the enriched alert to the elasticsearch
@@ -870,6 +936,7 @@ def batch_enrich_alerts(
         except Exception:
             logger.exception("Failed to push alerts to elasticsearch")
             pass
+
         # use pusher to push the enriched alert to the client
         pusher_client = get_pusher_client()
         if pusher_client:
@@ -884,9 +951,10 @@ def batch_enrich_alerts(
             except Exception:
                 logger.exception("Failed to tell client to poll alerts")
                 pass
+
         logger.info(
             "Alerts batch enriched successfully",
-            extra={"fingerprints": enrich_data.fingerprints, "tenant_id": tenant_id},
+            extra={"fingerprints": fingerprints, "tenant_id": tenant_id},
         )
 
         if should_run_workflow:
@@ -910,7 +978,9 @@ def batch_enrich_alerts(
                     session.commit()
 
         return {"status": "ok"}
-
+    except HTTPException:
+        # Re-raise HTTP exceptions
+        raise
     except Exception as e:
         logger.exception("Failed to enrich alerts batch", extra={"error": str(e)})
         return {"status": "failed"}
diff --git a/tests/test_batch_enrich_cel.py b/tests/test_batch_enrich_cel.py
new file mode 100644
index 0000000000..43192565f9
--- /dev/null
+++ b/tests/test_batch_enrich_cel.py
@@ -0,0 +1,477 @@
+import time
+from datetime import datetime
+
+import pytest
+
+from keep.api.models.alert import AlertStatus
+from tests.fixtures.client import client, setup_api_key, test_app  # noqa
+
+
+@pytest.mark.parametrize("test_app", ["NO_AUTH"], indirect=True)
+@pytest.mark.parametrize("elastic_client", [False], indirect=True)
+def test_batch_enrich_cel_basic(
+    db_session, client, test_app, create_alert, elastic_client
+):
+    """Test basic batch enrichment with a simple CEL expression (name matching)."""
+    # Create test alerts with specific names
+    create_alert(
+        "alert-cpu-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {"name": "CPU Overload Alert", "severity": "critical"},
+    )
+    create_alert(
+        "alert-memory-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {"name": "Memory Usage Alert", "severity": "warning"},
+    )
+    create_alert(
+        "alert-disk-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {"name": "Disk Space Alert", "severity": "warning"},
+    )
+
+    # Enrich alerts with name containing "CPU" via CEL
+    response = client.post(
+        "/alerts/batch_enrich",
+        headers={"x-api-key": "some-key"},
+        json={
+            "cel": "name.contains('CPU')",
+            "enrichments": {
+                "status": "acknowledged",
+                "note": "CPU issue being investigated",
+            },
+        },
+    )
+
+    assert response.status_code == 200
+    result = response.json()
+    assert result["status"] == "ok"
+
+    time.sleep(1)  # Allow time for async processing
+
+    # Get all alerts and verify only CPU alert was enriched
+    response = client.get(
+        "/preset/feed/alerts",
+        headers={"x-api-key": "some-key"},
+    )
+    alerts = response.json()
+
+    assert len(alerts) == 3
+
+    cpu_alert = next(a for a in alerts if a["name"] == "CPU Overload Alert")
+    memory_alert = next(a for a in alerts if a["name"] == "Memory Usage Alert")
+    disk_alert = next(a for a in alerts if a["name"] == "Disk Space Alert")
+
+    assert cpu_alert["status"] == "acknowledged"
+    assert cpu_alert["note"] == "CPU issue being investigated"
+    assert memory_alert["status"] == "firing"
+    assert disk_alert["status"] == "firing"
+
+
+@pytest.mark.parametrize("test_app", ["NO_AUTH"], indirect=True)
+@pytest.mark.parametrize("elastic_client", [False], indirect=True)
+def test_batch_enrich_cel_severity(
+    db_session, client, test_app, create_alert, elastic_client
+):
+    """Test batch enrichment with CEL expression filtering by severity."""
+    # Create test alerts with different severities
+    create_alert(
+        "alert-critical-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {"name": "Critical Service Down", "severity": "critical"},
+    )
+    create_alert(
+        "alert-warning-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {"name": "Warning Alert", "severity": "warning"},
+    )
+    create_alert(
+        "alert-warning-2",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {"name": "Another Warning", "severity": "warning"},
+    )
+
+    # Enrich all warning alerts
+    response = client.post(
+        "/alerts/batch_enrich",
+        headers={"x-api-key": "some-key"},
+        json={
+            "cel": "severity == 'warning'",
+            "enrichments": {
+                "status": "suppressed",
+                "note": "Low priority alerts suppressed",
+            },
+        },
+    )
+
+    assert response.status_code == 200
+    result = response.json()
+    assert result["status"] == "ok"
+
+    time.sleep(1)  # Allow time for async processing
+
+    # Verify only warning alerts were suppressed
+    response = client.get(
+        "/preset/feed/alerts",
+        headers={"x-api-key": "some-key"},
+    )
+    alerts = response.json()
+
+    assert len(alerts) == 3
+
+    critical_alert = next(a for a in alerts if a["severity"] == "critical")
+    warning_alerts = [a for a in alerts if a["severity"] == "warning"]
+
+    assert critical_alert["status"] == "firing"
+    assert all(a["status"] == "suppressed" for a in warning_alerts)
+    assert all(a["note"] == "Low priority alerts suppressed" for a in warning_alerts)
+
+
+@pytest.mark.parametrize("test_app", ["NO_AUTH"], indirect=True)
+@pytest.mark.parametrize("elastic_client", [False], indirect=True)
+def test_batch_enrich_cel_labels(
+    db_session, client, test_app, create_alert, elastic_client
+):
+    """Test batch enrichment with CEL expression filtering by labels."""
+    # Create test alerts with different labels
+    create_alert(
+        "alert-region1-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {
+            "name": "Region 1 Alert",
+            "severity": "critical",
+            "labels": {"region": "us-east-1", "service": "api"},
+        },
+    )
+    create_alert(
+        "alert-region1-2",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {
+            "name": "Region 1 Service Alert",
+            "severity": "warning",
+            "labels": {"region": "us-east-1", "service": "database"},
+        },
+    )
+    create_alert(
+        "alert-region2-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {
+            "name": "Region 2 Alert",
+            "severity": "critical",
+            "labels": {"region": "us-west-1", "service": "api"},
+        },
+    )
+
+    # Enrich alerts from us-east-1 region
+    response = client.post(
+        "/alerts/batch_enrich",
+        headers={"x-api-key": "some-key"},
+        json={
+            "cel": "labels.region == 'us-east-1'",
+            "enrichments": {
+                "status": "acknowledged",
+                "assignee": "east-team@example.com",
+            },
+        },
+    )
+
+    assert response.status_code == 200
+    result = response.json()
+    assert result["status"] == "ok"
+
+    time.sleep(1)  # Allow time for async processing
+
+    # Verify correct alerts were enriched
+    response = client.get(
+        "/preset/feed/alerts",
+        headers={"x-api-key": "some-key"},
+    )
+    alerts = response.json()
+
+    assert len(alerts) == 3
+
+    east_alerts = [
+        a for a in alerts if a.get("labels", {}).get("region") == "us-east-1"
+    ]
+    west_alerts = [
+        a for a in alerts if a.get("labels", {}).get("region") == "us-west-1"
+    ]
+
+    assert all(a["status"] == "acknowledged" for a in east_alerts)
+    assert all(a["assignee"] == "east-team@example.com" for a in east_alerts)
+    assert all(a["status"] == "firing" for a in west_alerts)
+
+
+@pytest.mark.parametrize("test_app", ["NO_AUTH"], indirect=True)
+@pytest.mark.parametrize("elastic_client", [False], indirect=True)
+def test_batch_enrich_cel_complex_expression(
+    db_session, client, test_app, create_alert, elastic_client
+):
+    """Test batch enrichment with a complex CEL expression combining multiple conditions."""
+    # Create test alerts with various properties
+    create_alert(
+        "alert-prod-critical-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {
+            "name": "Production Critical Alert",
+            "severity": "critical",
+            "environment": "production",
+            "service": "api",
+        },
+    )
+    create_alert(
+        "alert-prod-warning-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {
+            "name": "Production Warning Alert",
+            "severity": "warning",
+            "environment": "production",
+            "service": "api",
+        },
+    )
+    create_alert(
+        "alert-staging-critical-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {
+            "name": "Staging Critical Alert",
+            "severity": "critical",
+            "environment": "staging",
+            "service": "api",
+        },
+    )
+    create_alert(
+        "alert-prod-critical-2",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {
+            "name": "Production Critical DB Alert",
+            "severity": "critical",
+            "environment": "production",
+            "service": "database",
+        },
+    )
+
+    # Enrich critical production API alerts
+    response = client.post(
+        "/alerts/batch_enrich",
+        headers={"x-api-key": "some-key"},
+        json={
+            "cel": "severity == 'critical' && environment == 'production' && service == 'api'",
+            "enrichments": {
+                "status": "acknowledged",
+                "note": "Critical API issue - investigating",
+                "assignee": "api-team@example.com",
+            },
+        },
+    )
+
+    assert response.status_code == 200
+    result = response.json()
+    assert result["status"] == "ok"
+
+    time.sleep(1)  # Allow time for async processing
+
+    # Verify only the matching alert was enriched
+    response = client.get(
+        "/preset/feed/alerts",
+        headers={"x-api-key": "some-key"},
+    )
+    alerts = response.json()
+
+    assert len(alerts) == 4
+
+    # Find the production critical API alert
+    prod_critical_api = next(
+        a
+        for a in alerts
+        if a["environment"] == "production"
+        and a["severity"] == "critical"
+        and a["service"] == "api"
+    )
+
+    # Check it was properly enriched
+    assert prod_critical_api["status"] == "acknowledged"
+    assert prod_critical_api["note"] == "Critical API issue - investigating"
+    assert prod_critical_api["assignee"] == "api-team@example.com"
+
+    # Check other alerts were not enriched
+    other_alerts = [
+        a
+        for a in alerts
+        if not (
+            a["environment"] == "production"
+            and a["severity"] == "critical"
+            and a["service"] == "api"
+        )
+    ]
+    assert all(a["status"] == "firing" for a in other_alerts)
+    assert all(not a.get("assignee") for a in other_alerts)
+
+
+@pytest.mark.parametrize("test_app", ["NO_AUTH"], indirect=True)
+@pytest.mark.parametrize("elastic_client", [False], indirect=True)
+def test_batch_enrich_cel_no_matching_alerts(
+    db_session, client, test_app, create_alert, elastic_client
+):
+    """Test batch enrichment when no alerts match the CEL expression."""
+    # Create some alerts
+    create_alert(
+        "alert-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {"name": "Test Alert 1", "severity": "critical"},
+    )
+    create_alert(
+        "alert-2",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {"name": "Test Alert 2", "severity": "warning"},
+    )
+
+    # Use a CEL expression that won't match any alerts
+    response = client.post(
+        "/alerts/batch_enrich",
+        headers={"x-api-key": "some-key"},
+        json={
+            "cel": "name.contains('NonExistentString')",
+            "enrichments": {
+                "status": "acknowledged",
+            },
+        },
+    )
+
+    assert response.status_code == 200
+    result = response.json()
+    assert result["status"] == "ok"
+    assert "message" in result
+    assert "No alerts matched the query" in result["message"]
+
+    # Verify no alerts were changed
+    response = client.get(
+        "/preset/feed/alerts",
+        headers={"x-api-key": "some-key"},
+    )
+    alerts = response.json()
+
+    assert len(alerts) == 2
+    assert all(a["status"] == "firing" for a in alerts)
+
+
+@pytest.mark.parametrize("test_app", ["NO_AUTH"], indirect=True)
+@pytest.mark.parametrize("elastic_client", [False], indirect=True)
+def test_batch_enrich_cel_invalid_expression(
+    db_session, client, test_app, create_alert, elastic_client
+):
+    """Test batch enrichment with an invalid CEL expression."""
+    # Create an alert
+    create_alert(
+        "alert-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {"name": "Test Alert", "severity": "critical"},
+    )
+
+    # Use an invalid CEL expression
+    response = client.post(
+        "/alerts/batch_enrich",
+        headers={"x-api-key": "some-key"},
+        json={
+            "cel": "invalid.syntax &&& !!!",
+            "enrichments": {
+                "status": "acknowledged",
+            },
+        },
+    )
+
+    # Should return a 400 error
+    assert response.status_code == 400
+    assert "Error parsing CEL expression" in response.json()["detail"]
+
+    # Verify no alerts were changed
+    response = client.get(
+        "/preset/feed/alerts",
+        headers={"x-api-key": "some-key"},
+    )
+    alerts = response.json()
+
+    assert len(alerts) == 1
+    assert alerts[0]["status"] == "firing"
+
+
+@pytest.mark.parametrize("test_app", ["NO_AUTH"], indirect=True)
+@pytest.mark.parametrize("elastic_client", [False], indirect=True)
+def test_batch_enrich_cel_dispose_on_new_alert(
+    db_session, client, test_app, create_alert, elastic_client
+):
+    """Test batch enrichment with dispose_on_new_alert parameter."""
+    # Create an alert
+    create_alert(
+        "alert-test-1",
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {"name": "Test Alert", "severity": "critical"},
+    )
+
+    # Enrich the alert with dispose_on_new_alert=True
+    response = client.post(
+        "/alerts/batch_enrich?dispose_on_new_alert=true",
+        headers={"x-api-key": "some-key"},
+        json={
+            "cel": "name.contains('Test')",
+            "enrichments": {
+                "status": "resolved",
+                "note": "Temporary resolution note",
+            },
+        },
+    )
+
+    assert response.status_code == 200
+    result = response.json()
+    assert result["status"] == "ok"
+
+    time.sleep(1)  # Allow time for async processing
+
+    # Verify the alert was enriched
+    response = client.get(
+        "/preset/feed/alerts",
+        headers={"x-api-key": "some-key"},
+    )
+    alerts = response.json()
+
+    assert len(alerts) == 1
+    assert alerts[0]["status"] == "resolved"
+    assert alerts[0]["note"] == "Temporary resolution note"
+
+    # Create a new alert with the same fingerprint to simulate a new occurrence
+    create_alert(
+        "alert-test-1",  # Same fingerprint
+        AlertStatus.FIRING,
+        datetime.utcnow(),
+        {"name": "Test Alert", "severity": "critical"},
+    )
+
+    time.sleep(1)  # Allow time for processing
+
+    # Verify the new alert has the disposable enrichment
+    response = client.get(
+        "/preset/feed/alerts",
+        headers={"x-api-key": "some-key"},
+    )
+    alerts = response.json()
+
+    # This assertion checks the behavior described in the issue
+    # Note: Based on the reported issue, this might actually show the enrichment persisting
+    # which is the behavior being reported as problematic
+    assert len(alerts) == 1
diff --git a/tests/test_enrichments.py b/tests/test_enrichments.py
index 649b76c8c1..9c05dbb4b2 100644
--- a/tests/test_enrichments.py
+++ b/tests/test_enrichments.py
@@ -431,7 +431,9 @@ def test_mapping_rule_with_elastic(mock_session, mock_alert_dto, setup_alerts):
 
 
 @pytest.mark.parametrize("test_app", ["NO_AUTH"], indirect=True)
-def test_enrichment_with_elastic(db_session, client, test_app, mock_alert_dto, elastic_client):
+def test_enrichment_with_elastic(
+    db_session, client, test_app, mock_alert_dto, elastic_client
+):
     # add some rule
     rule = MappingRule(
         id=1,
@@ -881,7 +883,8 @@ def test_batch_enrichment(db_session, client, test_app, create_alert, elastic_cl
     )
 
     assert response.status_code == 200
-    assert response.json() == {"status":"ok"}
+    response_data = response.json()
+    assert response_data["status"] == "ok"
 
     time.sleep(1)
 


Code review comments:
diff_hunk: @@ -805,6 +804,78 @@ def batch_enrich_alerts(
     ):
         enrich_data.enrichments["status"] = AlertStatus.SUPPRESSED.value
 
+    if not enrich_data.fingerprints and not enrich_data.cel:
+        raise HTTPException(
+            status_code=400, detail="Either fingerprints or cel must be provided"
+        )
+
+    if enrich_data.fingerprints and enrich_data.cel:
+        raise HTTPException(
+            status_code=400, detail="Either fingerprints or cel can be provided at once"
+        )
+
+    # If CEL is provided, use it to find matching alerts
+    if enrich_data.cel:
+        logger.info(
+            "Enriching alerts by CEL query",
+            extra={
+                "cel": enrich_data.cel,
+                "tenant_id": tenant_id,
+            },
+        )
+
+        try:
+            db_alerts, total_count = query_last_alerts(
+                tenant_id=tenant_id,
+                query=QueryDto(
+                    cel=enrich_data.cel,
+                    limit=enrich_data.limit,
+                    offset=enrich_data.offset,
path: keep/api/routes/alerts.py
text: @user1:
```suggestion
                    offset=0,
```

@author:
Deleted at https://github.com/keephq/keep/pull/4568/commits/524d00c72d1ec0b5f985577dbb7f21a67a0dc34c.

diff_hunk: @@ -805,6 +804,78 @@ def batch_enrich_alerts(
     ):
         enrich_data.enrichments["status"] = AlertStatus.SUPPRESSED.value
 
+    if not enrich_data.fingerprints and not enrich_data.cel:
+        raise HTTPException(
+            status_code=400, detail="Either fingerprints or cel must be provided"
+        )
+
+    if enrich_data.fingerprints and enrich_data.cel:
+        raise HTTPException(
+            status_code=400, detail="Either fingerprints or cel can be provided at once"
+        )
+
+    # If CEL is provided, use it to find matching alerts
+    if enrich_data.cel:
+        logger.info(
+            "Enriching alerts by CEL query",
+            extra={
+                "cel": enrich_data.cel,
+                "tenant_id": tenant_id,
+            },
+        )
+
+        try:
+            db_alerts, total_count = query_last_alerts(
+                tenant_id=tenant_id,
+                query=QueryDto(
+                    cel=enrich_data.cel,
+                    limit=enrich_data.limit,
path: keep/api/routes/alerts.py
text: @user1:
```suggestion
                    limit=1000,
```

@author:
Deleted at https://github.com/keephq/keep/pull/4568/commits/524d00c72d1ec0b5f985577dbb7f21a67a0dc34c.

diff_hunk: @@ -805,6 +804,78 @@ def batch_enrich_alerts(
     ):
         enrich_data.enrichments["status"] = AlertStatus.SUPPRESSED.value
 
+    if not enrich_data.fingerprints and not enrich_data.cel:
+        raise HTTPException(
+            status_code=400, detail="Either fingerprints or cel must be provided"
+        )
+
+    if enrich_data.fingerprints and enrich_data.cel:
+        raise HTTPException(
+            status_code=400, detail="Either fingerprints or cel can be provided at once"
+        )
+
+    # If CEL is provided, use it to find matching alerts
+    if enrich_data.cel:
+        logger.info(
+            "Enriching alerts by CEL query",
+            extra={
+                "cel": enrich_data.cel,
+                "tenant_id": tenant_id,
+            },
+        )
+
+        try:
+            db_alerts, total_count = query_last_alerts(
+                tenant_id=tenant_id,
+                query=QueryDto(
+                    cel=enrich_data.cel,
+                    limit=enrich_data.limit,
path: keep/api/routes/alerts.py
text: @user1:
Just thinking out loud:
Shouldn't the backend handle this internally?
It seems more appropriate for the backend to enrich all alerts matching the provided CEL expression, rather than offloading that responsibility to the client.
Also, having limit and offset parameters for enrichment feels a bit confusing to me.
What do you think?
Please let me know if I missed something.

@author:
Hi @user1, can you provide more explanation about the statement `Shouldn't the backend handle this internally? It seems more appropriate for the backend to enrich all alerts matching the provided CEL expression, rather than offloading that responsibility to the client.` since I don't get what is the client and backend in our context. Sorry, just a beginner to the software development journey :D 

I add the option to provide limit and offset because the `QueryDTO` had it, also if you don't provide, the default value of limit will be `1000` and offset is `0`. What about either we add option that limit can be `-1` which is no limitation or change the default to no limitation instead of `1000`.

@user1:
What I meant is that if a client (e.g., someone triggering the backend endpoint—like Keep UI, for example) initiates a batch enrichment with a CEL expression like name.contains('CPU') and sets limit to 1000, but there are actually 1423 matching alerts—what happens to the remaining 423?

My point is that the backend should ideally take care of this internally. If the total number of matching alerts exceeds the limit, the backend should handle pagination or batching automatically and ensure that all matching alerts get enriched, regardless of how many there are.

From my perspective, the client should only need to provide the CEL expression. limit and offset should be managed internally by the backend.

@author:
Thanks @user1, I understand it.

Just one more question: Is the reason for setting the default limit to 1000 to avoid returning a large amount of data?

If so, should we consider implementing a technique to fetch all data in batches sequentially and then process enriching each batch, instead of imposing a hard limit of only `1000` ? For example if total count is `10000`, then we may have to process each batch `1000` one by one.

@user1:
Yes, I think it's a good way to go

@author:
Thanks @user1, for simplicity, I think we can go ahead and get this PR reviewed and merged first. We can introduce the improvements in future PRs. 😄

@user1:
I don't mind merging it.
Left a final comment.

diff_hunk: @@ -367,7 +367,10 @@ class EnrichAlertRequestBody(BaseModel):
 
 class BatchEnrichAlertRequestBody(BaseModel):
     enrichments: dict[str, str]
-    fingerprints: list[str]
+    fingerprints: Optional[list[str]] = None
+    cel: Optional[str] = None
+    limit: int = 1000
+    offset: int = 0
path: keep/api/models/alert.py
text: @user1:
```suggestion
```

@author:
Deleted at https://github.com/keephq/keep/pull/4568/commits/524d00c72d1ec0b5f985577dbb7f21a67a0dc34c.

diff_hunk: @@ -367,7 +367,10 @@ class EnrichAlertRequestBody(BaseModel):
 
 class BatchEnrichAlertRequestBody(BaseModel):
     enrichments: dict[str, str]
-    fingerprints: list[str]
+    fingerprints: Optional[list[str]] = None
+    cel: Optional[str] = None
+    limit: int = 1000
path: keep/api/models/alert.py
text: @user1:
Let's maybe remove these limit/offset parameters from batch enrichment request and hardcode limit:1000 offset:0
in the place where it's used now. I think no need to make limit and offset being part of public api especially because it's gonna be change in future.

@author:
Deleted at https://github.com/keephq/keep/pull/4568/commits/524d00c72d1ec0b5f985577dbb7f21a67a0dc34c.