Fix: Standardize URL formation for dev/staging regions across Chronicle SDK

CHANGELOG.md

@@ -5,7 +5,16 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-
+## [0.34.1] - 2026-01-29
+### Updated
+- Following methods for streamlined URL formation/construction
+  - `ingest_udm`
+  - `translate_nl_to_udm`
+  - `create_conversation`
+  - `opt_in_to_gemini`
+  - `query_gemini`
+  - `translate_nl_to_udm`
+  - `run_rule_test`
 
 ## [0.34.0] - 2026-01-12
 ### Added

api_module_mapping.md

@@ -4,6 +4,13 @@ Following shows mapping between SecOps [REST Resource](https://cloud.google.com/
 
 **Note:** All the REST resources mentioned have suffix `projects.locations.instances`.
 
+## Implementation Statistics
+
+- **v1:** 17 endpoints implemented
+- **v1alpha:** 113 endpoints implemented
+
+## Endpoint Mapping
+
 | REST Resource                                                                  | Version | secops-wrapper module                                                                                             | CLI Command                                    |
 |--------------------------------------------------------------------------------|---------|-------------------------------------------------------------------------------------------------------------------|------------------------------------------------|
 | dataAccessLabels.create                                                        | v1      |                                                                                                                   |                                                |

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "secops"
-version = "0.34.0"
+version = "0.34.1"
 description = "Python SDK for wrapping the Google SecOps API for common use cases"
 readme = "README.md"
 requires-python = ">=3.10"

src/secops/chronicle/data_export.py

@@ -67,10 +67,7 @@ def _get_formatted_log_type(client, log_type: str) -> str:
         The formatted log type
     """
     if "/" not in log_type:
-        return (
-            f"projects/{client.project_id}/locations/{client.region}/"
-            f"instances/{client.customer_id}/logTypes/{log_type}"
-        )
+        return f"{client.instance_id}/logTypes/{log_type}"
 
     return log_type
 

src/secops/chronicle/gemini.py

@@ -327,10 +327,7 @@ def create_conversation(client, display_name: str = "New chat") -> str:
     Raises:
         APIError: If the API request fails
     """
-    url = (
-        f"{client.base_url}/projects/{client.project_id}/locations/"
-        f"{client.region}/instances/{client.customer_id}/users/me/conversations"
-    )
+    url = f"{client.base_url}/{client.instance_id}/users/me/conversations"
 
     # Include the required request body with displayName
     payload = {"displayName": display_name}
@@ -369,10 +366,7 @@ def opt_in_to_gemini(client) -> bool:
         APIError: If the API request fails (except for permission errors)
     """
     # Construct the URL for updating the user's preference set
-    url = (
-        f"{client.base_url}/projects/{client.project_id}/locations/"
-        f"{client.region}/instances/{client.customer_id}/users/me/preferenceSet"
-    )
+    url = f"{client.base_url}/{client.instance_id}/users/me/preferenceSet"
 
     # Set up the request body to enable Duet AI chat
     payload = {"ui_preferences": {"enable_duet_ai_chat": True}}
@@ -447,8 +441,7 @@ def query_gemini(
             conversation_id = create_conversation(client)
 
         url = (
-            f"{client.base_url}/projects/{client.project_id}/locations/"
-            f"{client.region}/instances/{client.customer_id}/users/me/"
+            f"{client.base_url}/{client.instance_id}/users/me/"
             f"conversations/{conversation_id}/messages"
         )
 

src/secops/chronicle/log_ingest.py

@@ -24,6 +24,7 @@
 from typing import Any
 
 from secops.chronicle.log_types import is_valid_log_type
+from secops.chronicle.models import APIVersion
 from secops.exceptions import APIError
 
 # Forward declaration for type hinting to avoid circular import
@@ -1010,14 +1011,9 @@ def ingest_udm(
         if add_missing_ids and "id" not in event["metadata"]:
             event["metadata"]["id"] = str(uuid.uuid4())
 
-    # Prepare the request
-    parent = (
-        f"projects/{client.project_id}/locations/{client.region}"
-        f"/instances/{client.customer_id}"
-    )
     url = (
-        f"https://{client.region}-chronicle.googleapis.com/v1alpha/"
-        f"{parent}/events:import"
+        f"{client.base_url(APIVersion.V1ALPHA)}/{client.instance_id}"
+        f"/events:import"
     )
 
     # Format the request body

src/secops/chronicle/nl_search.py

@@ -18,6 +18,7 @@
 from datetime import datetime
 from typing import Any
 
+from secops.chronicle.models import APIVersion
 from secops.exceptions import APIError
 
 
@@ -40,9 +41,8 @@ def translate_nl_to_udm(client, text: str) -> str:
     wait_time = 5  # seconds, will double with each retry
 
     url = (
-        f"https://{client.region}-chronicle.googleapis.com/v1alpha/projects"
-        f"/{client.project_id}/locations/{client.region}/instances"
-        f"/{client.customer_id}:translateUdmQuery"
+        f"{client.base_url(APIVersion.V1ALPHA)}/{client.instance_id}"
+        f":translateUdmQuery"
     )
 
     payload = {"text": text}

src/secops/chronicle/rule.py

@@ -439,11 +439,7 @@ def run_rule_test(
     end_time_str = end_time.strftime("%Y-%m-%dT%H:%M:%SZ")
 
     # Fix: Use the full path for the legacy API endpoint
-    url = (
-        f"{client.base_url}/projects/{client.project_id}/locations"
-        f"/{client.region}/instances/{client.customer_id}"
-        "/legacy:legacyRunTestRule"
-    )
+    url = f"{client.base_url}/{client.instance_id}/legacy:legacyRunTestRule"
 
     body = {
         "ruleText": rule_text,

tests/chronicle/test_gemini.py

@@ -36,6 +36,9 @@ def mock_chronicle_client():
     client.customer_id = "test-customer"
     client.region = "us"
     client.base_url = "https://us-chronicle.googleapis.com/v1alpha"
+    client.instance_id = (
+        "projects/test-project/locations/us/instances/test-customer"
+    )
     client.session = Mock()
     return client
 
@@ -281,10 +284,11 @@ def test_create_conversation_success(mock_chronicle_client):
     call_args = mock_chronicle_client.session.post.call_args
 
     # Check URL
-    assert (
-        call_args[0][0]
-        == "https://us-chronicle.googleapis.com/v1alpha/projects/test-project/locations/us/instances/test-customer/users/me/conversations"
+    expected_url = (
+        f"{mock_chronicle_client.base_url}/"
+        f"{mock_chronicle_client.instance_id}/users/me/conversations"
     )
+    assert call_args[0][0] == expected_url
 
     # Check payload
     assert call_args[1]["json"] == {"displayName": "New chat"}
@@ -491,7 +495,7 @@ def test_query_gemini_auto_opt_in(mock_chronicle_client, sample_gemini_response)
             '{"error":{"message":"users must opt-in before using Gemini"}}'
         )
         first_response.raise_for_status.side_effect = requests.exceptions.HTTPError(
-            "400 Client Error", response=first_response
+                "400 Client Error", response=first_response
         )
 
         # Second request succeeds
@@ -539,7 +543,7 @@ def test_query_gemini_opt_in_flag(mock_chronicle_client, sample_gemini_response)
             '{"error":{"message":"users must opt-in before using Gemini"}}'
         )
         opt_in_error.raise_for_status.side_effect = requests.exceptions.HTTPError(
-            "400 Client Error", response=opt_in_error
+                "400 Client Error", response=opt_in_error
         )
 
         success_response = Mock()

tests/chronicle/test_nl_search.py

@@ -29,6 +29,12 @@ def mock_client():
     client.region = "us"
     client.project_id = "test-project"
     client.customer_id = "test-customer-id"
+    client.base_url = MagicMock(
+        return_value="https://us-chronicle.googleapis.com/v1alpha"
+    )
+    client.instance_id = (
+        "projects/test-project/locations/us/instances/test-customer-id"
+    )
 
     # Mock session with response
     mock_response = MagicMock()
@@ -49,11 +55,11 @@ def test_translate_nl_to_udm_success(mock_client):
 
     # Check URL format
     url = call_args[0][0]
-    assert "us-chronicle.googleapis.com" in url
-    assert "/v1alpha/" in url
-    assert "test-project" in url
-    assert "test-customer-id" in url
-    assert ":translateUdmQuery" in url
+    expected_url = (
+        f"{mock_client.base_url.return_value}/"
+        f"{mock_client.instance_id}:translateUdmQuery"
+    )
+    assert url == expected_url
 
     # Check payload
     payload = call_args[1]["json"]
@@ -72,7 +78,9 @@ def test_translate_nl_to_udm_error_response(mock_client):
     mock_client.session.post.return_value = mock_response
 
     # Test error handling
-    with pytest.raises(APIError, match="Chronicle API request failed: Invalid request"):
+    with pytest.raises(
+        APIError, match="Chronicle API request failed: Invalid request"
+    ):
         translate_nl_to_udm(mock_client, "invalid query")
 
 
@@ -87,7 +95,9 @@ def test_translate_nl_to_udm_no_valid_query(mock_client):
     mock_client.session.post.return_value = mock_response
 
     # Test error handling for no valid query
-    with pytest.raises(APIError, match="Sorry, no valid query could be generated"):
+    with pytest.raises(
+        APIError, match="Sorry, no valid query could be generated"
+    ):
         translate_nl_to_udm(mock_client, "nonsensical query")
 
 
@@ -103,7 +113,9 @@ def test_nl_search(mock_translate, mock_client):
     end_time = datetime.now(timezone.utc)
 
     # Call the function
-    result = nl_search(mock_client, "show me ip addresses", start_time, end_time)
+    result = nl_search(
+        mock_client, "show me ip addresses", start_time, end_time
+    )
 
     # Verify translate_nl_to_udm was called
     mock_translate.assert_called_once_with(mock_client, "show me ip addresses")
@@ -123,14 +135,18 @@ def test_nl_search(mock_translate, mock_client):
 def test_nl_search_translation_error(mock_translate, mock_client):
     """Test error handling when translation fails."""
     # Set up translation to raise an error
-    mock_translate.side_effect = APIError("Sorry, no valid query could be generated")
+    mock_translate.side_effect = APIError(
+        "Sorry, no valid query could be generated"
+    )
 
     # Define test parameters
     start_time = datetime.now(timezone.utc) - timedelta(hours=24)
     end_time = datetime.now(timezone.utc)
 
     # Test error handling
-    with pytest.raises(APIError, match="Sorry, no valid query could be generated"):
+    with pytest.raises(
+        APIError, match="Sorry, no valid query could be generated"
+    ):
         nl_search(mock_client, "invalid query", start_time, end_time)
 
     # Verify search_udm was not called
@@ -153,7 +169,9 @@ def test_chronicle_client_integration():
     assert (
         client_method is not None
     ), "translate_nl_to_udm method not found on ChronicleClient"
-    assert search_method is not None, "nl_search method not found on ChronicleClient"
+    assert (
+        search_method is not None
+    ), "nl_search method not found on ChronicleClient"
 
     # Additional check from the module import
     assert hasattr(ChronicleClient, "translate_nl_to_udm")
@@ -208,7 +226,9 @@ def test_nl_search_retry_429(mock_sleep, mock_translate, mock_client):
     end_time = datetime.now(timezone.utc)
 
     # Call the function
-    result = nl_search(mock_client, "show me ip addresses", start_time, end_time)
+    result = nl_search(
+        mock_client, "show me ip addresses", start_time, end_time
+    )
 
     # Verify translate_nl_to_udm was called at least once with correct arguments
     # We expect it to be called on each retry attempt
@@ -229,7 +249,9 @@ def test_nl_search_retry_429(mock_sleep, mock_translate, mock_client):
 
 @patch("secops.chronicle.nl_search.translate_nl_to_udm")
 @patch("time.sleep")  # Patch sleep to avoid waiting in tests
-def test_nl_search_max_retries_exceeded(mock_sleep, mock_translate, mock_client):
+def test_nl_search_max_retries_exceeded(
+    mock_sleep, mock_translate, mock_client
+):
     """Test that max retries are respected for 429 errors."""
     # Set up mock for translation
     mock_translate.return_value = 'ip != ""'