fix(connector) Use intake URL from file

fix

new version

remove extra space

run ruff changelog and version

run black

Add test

update

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.12.2] - 2024-03-26
+
+### Fixed
+
+- Use file for batch url computation
+
 ## [1.12.1] - 2024-03-22
 
 ### Fixed

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry]
 name = "sekoia-automation-sdk"
 
-version = "1.12.1"
+version = "1.12.2"
 description = "SDK to create Sekoia.io playbook modules"
 license = "MIT"
 readme = "README.md"

sekoia_automation/aio/connector.py

@@ -1,6 +1,5 @@
 """Contains connector with async version."""
 
-import os
 from abc import ABC
 from asyncio import AbstractEventLoop, get_event_loop
 from collections.abc import AsyncGenerator
@@ -154,8 +153,10 @@ async def push_data_to_intakes(
             list[str]:
         """
         self._last_events_time = datetime.utcnow()
-        intake_host = os.getenv("INTAKE_URL", self.configuration.intake_server)
-        batch_api = urljoin(intake_host, "batch")
+        if intake_server := self.configuration.intake_server:
+            batch_api = urljoin(intake_server, "batch")
+        else :
+            batch_api = urljoin(self.intake_url, "batch")
 
         result_ids = []
 

sekoia_automation/connector/__init__.py

@@ -1,4 +1,3 @@
-import os
 import uuid
 from abc import ABC
 from collections.abc import Generator, Sequence
@@ -36,7 +35,7 @@
 
 
 class DefaultConnectorConfiguration(BaseModel):
-    intake_server: str = "https://intake.sekoia.io"
+    intake_server: str | None = None
     intake_key: str
 
 
@@ -171,8 +170,10 @@ def push_events_to_intakes(
         # Reset the consecutive error count
         self._error_count = 0
         self._last_events_time = datetime.utcnow()
-        intake_host = os.getenv("INTAKE_URL", self.configuration.intake_server)
-        batch_api = urljoin(intake_host, "batch")
+        if intake_server := self.configuration.intake_server:
+            batch_api = urljoin(intake_server, "batch")
+        else :
+            batch_api = urljoin(self.intake_url, "batch")
 
         # Dict to collect event_ids for the API
         collect_ids: dict[int, list] = {}

sekoia_automation/module.py

@@ -322,6 +322,7 @@ class ModuleItem(ABC):
     CALLBACK_URL_FILE_NAME = "url_callback"
     SECRETS_URL_FILE_NAME = "url_secrets"
     LOGS_URL_FILE_NAME = "url_logs"
+    INTAKE_URL_FILE_NAME = "intake_url"
 
     name: str | None = None
     description: str | None = None
@@ -419,6 +420,10 @@ def secrets_url(self) -> str:
         except FileNotFoundError:
             return self.callback_url.replace("/callback", "/secrets")
 
+    @cached_property
+    def intake_url(self) -> str:
+        return self.module.load_config(self.INTAKE_URL_FILE_NAME)
+
     @property
     def _headers(self) -> dict[str, str]:
         return {"Authorization": f"Bearer {self.token}"}

tests/test_trigger.py

@@ -61,6 +61,13 @@ def test_secrets_url():
     mock.assert_called_with(trigger.SECRETS_URL_FILE_NAME)
 
 
+def test_intake_url():
+    trigger = DummyTrigger()
+    with patch.object(Module, "load_config", return_value="intake") as mock:
+        assert trigger.intake_url == "intake"
+    mock.assert_called_with(trigger.INTAKE_URL_FILE_NAME)
+
+
 def test_logs_url():
     trigger = DummyTrigger()
     with patch.object(Module, "load_config", return_value="logs") as mock: