fix(python): preserve dynamic admin endpoint

Author: josedab

streamline_sdk/_admin_http.py

@@ -12,7 +12,7 @@
 
 import asyncio
 import json
-from collections.abc import Collection
+from collections.abc import Callable, Collection
 from typing import Any
 
 from .exceptions import TopicError
@@ -34,17 +34,20 @@ class _AdminHttpTransport:
     ``urllib.request`` call executed on a worker thread otherwise.
     """
 
-    def __init__(self, base_url: str) -> None:
+    def __init__(self, base_url: str | Callable[[], str]) -> None:
         """Initialize the transport.
 
         Args:
-            base_url: Base URL of the Streamline HTTP REST API.
+            base_url: Base URL or a callable that resolves the current URL.
         """
         self._base_url = base_url
 
+    def _current_base_url(self) -> str:
+        return self._base_url() if callable(self._base_url) else self._base_url
+
     async def get(self, path: str) -> Any:
         """Make an HTTP GET request to the Streamline REST API."""
-        url = f"{self._base_url}{path}"
+        url = f"{self._current_base_url()}{path}"
 
         if HAS_AIOHTTP:
             async with aiohttp.ClientSession() as session:
@@ -69,7 +72,7 @@ def _sync_get():
 
     async def post(self, path: str, body: Any) -> Any:
         """Make an HTTP POST request to the Streamline REST API."""
-        url = f"{self._base_url}{path}"
+        url = f"{self._current_base_url()}{path}"
 
         if HAS_AIOHTTP:
             async with aiohttp.ClientSession() as session:
@@ -97,7 +100,7 @@ def _sync_post():
 
     async def delete(self, path: str) -> None:
         """Make an HTTP DELETE request to the Streamline REST API."""
-        url = f"{self._base_url}{path}"
+        url = f"{self._current_base_url()}{path}"
 
         if HAS_AIOHTTP:
             async with aiohttp.ClientSession() as session:

streamline_sdk/admin.py

@@ -236,7 +236,7 @@ def __init__(self, client_config: Any):
         self._client_config = client_config
         self._admin: AIOKafkaAdminClient | None = None
         self._started = False
-        self._http = _AdminHttpTransport(client_config.http_url)
+        self._http = _AdminHttpTransport(lambda: self._client_config.http_url)
 
     async def start(self) -> None:
         """Start the admin client."""

tests/test_admin_transport.py

@@ -575,3 +575,12 @@ async def _recording_to_thread(func: Any, *args: Any, **kwargs: Any) -> Any:
         await transport.delete("/v1/branches/exp-a")
 
         assert len(to_thread_calls) == 1
+
+
+def test_transport_resolves_dynamic_base_url():
+    current = {"url": "http://first:9094"}
+    transport = _AdminHttpTransport(lambda: current["url"])
+
+    assert transport._current_base_url() == "http://first:9094"
+    current["url"] = "http://second:9094"
+    assert transport._current_base_url() == "http://second:9094"