From 59a13438968885bc056dd5b120b07e68230f16a4 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 5 Mar 2026 01:52:59 +0000 Subject: [PATCH 1/2] Fix: Parse nested targets within label_selector load balancer targets When a load balancer has a label_selector type target, the API returns a nested "targets" array containing the resolved individual server targets with their health statuses. This data was previously discarded. - Add `targets` parameter to `LoadBalancerTarget.__init__` in domain.py - Parse nested targets in `BoundLoadBalancer.__init__` for label_selector targets, creating `LoadBalancerTarget` objects with server, health_status, type, and use_private_ip fields - Add test fixture and test case for nested target parsing https://claude.ai/code/session_018oDf81V2LsgKcFheq7KSRp --- hcloud/load_balancers/client.py | 27 ++++++++++++++++++++++++ hcloud/load_balancers/domain.py | 3 +++ tests/unit/load_balancers/conftest.py | 18 +++++++++++++++- tests/unit/load_balancers/test_client.py | 23 ++++++++++++++++++++ 4 files changed, 70 insertions(+), 1 deletion(-) diff --git a/hcloud/load_balancers/client.py b/hcloud/load_balancers/client.py index 1986905b..ca73cd9b 100644 --- a/hcloud/load_balancers/client.py +++ b/hcloud/load_balancers/client.py @@ -108,6 +108,33 @@ def __init__( selector=target["label_selector"]["selector"] ) tmp_target.use_private_ip = target["use_private_ip"] + nested_targets = target.get("targets", []) + if nested_targets: + tmp_nested = [] + for nested in nested_targets: + nested_target = LoadBalancerTarget(type=nested["type"]) + if nested["type"] == "server": + nested_target.server = BoundServer( + client._parent.servers, + data=nested["server"], + complete=False, + ) + elif nested["type"] == "ip": + nested_target.ip = LoadBalancerTargetIP( + ip=nested["ip"]["ip"] + ) + nested_target.use_private_ip = nested.get("use_private_ip") + nested_health_status = nested.get("health_status") + if nested_health_status is not None: + nested_target.health_status = [ + LoadBalancerTargetHealthStatus( + listen_port=hs["listen_port"], + status=hs["status"], + ) + for hs in nested_health_status + ] + tmp_nested.append(nested_target) + tmp_target.targets = tmp_nested elif target["type"] == "ip": tmp_target.ip = LoadBalancerTargetIP(ip=target["ip"]["ip"]) diff --git a/hcloud/load_balancers/domain.py b/hcloud/load_balancers/domain.py index c02e1100..8ad5d28d 100644 --- a/hcloud/load_balancers/domain.py +++ b/hcloud/load_balancers/domain.py @@ -420,6 +420,7 @@ class LoadBalancerTarget(BaseDomain): "ip", "use_private_ip", "health_status", + "targets", ) __slots__ = __api_properties__ @@ -431,6 +432,7 @@ def __init__( ip: LoadBalancerTargetIP | None = None, use_private_ip: bool | None = None, health_status: list[LoadBalancerTargetHealthStatus] | None = None, + targets: list[LoadBalancerTarget] | None = None, ): self.type = type self.server = server @@ -438,6 +440,7 @@ def __init__( self.ip = ip self.use_private_ip = use_private_ip self.health_status = health_status + self.targets = targets def to_payload(self) -> dict[str, Any]: """ diff --git a/tests/unit/load_balancers/conftest.py b/tests/unit/load_balancers/conftest.py index f19508ea..e505fc1d 100644 --- a/tests/unit/load_balancers/conftest.py +++ b/tests/unit/load_balancers/conftest.py @@ -86,7 +86,23 @@ def response_load_balancer(): "health_status": [{"listen_port": 443, "status": "healthy"}], "label_selector": None, "use_private_ip": False, - } + }, + { + "type": "label_selector", + "label_selector": {"selector": "eu"}, + "use_private_ip": True, + "targets": [ + { + "type": "server", + "server": {"id": 105054278}, + "use_private_ip": True, + "health_status": [ + {"listen_port": 443, "status": "healthy"}, + {"listen_port": 3000, "status": "healthy"}, + ], + } + ], + }, ], "algorithm": {"type": "round_robin"}, } diff --git a/tests/unit/load_balancers/test_client.py b/tests/unit/load_balancers/test_client.py index 7ac3290c..20fe6eb7 100644 --- a/tests/unit/load_balancers/test_client.py +++ b/tests/unit/load_balancers/test_client.py @@ -60,6 +60,29 @@ def test_init(self, response_load_balancer): assert bound_load_balancer.id == 4711 assert bound_load_balancer.name == "Web Frontend" + def test_init_label_selector_nested_targets(self, response_load_balancer): + bound_load_balancer = BoundLoadBalancer( + client=mock.MagicMock(), data=response_load_balancer["load_balancer"] + ) + + label_selector_target = bound_load_balancer.targets[1] + assert label_selector_target.type == "label_selector" + assert label_selector_target.label_selector.selector == "eu" + assert label_selector_target.use_private_ip is True + assert label_selector_target.targets is not None + assert len(label_selector_target.targets) == 1 + + nested = label_selector_target.targets[0] + assert nested.type == "server" + assert nested.server.id == 105054278 + assert nested.use_private_ip is True + assert nested.health_status is not None + assert len(nested.health_status) == 2 + assert nested.health_status[0].listen_port == 443 + assert nested.health_status[0].status == "healthy" + assert nested.health_status[1].listen_port == 3000 + assert nested.health_status[1].status == "healthy" + class TestLoadBalancerslient: @pytest.fixture() From cd3e6aa6958db3d100850425da61cbc1e6257c61 Mon Sep 17 00:00:00 2001 From: jo Date: Thu, 5 Mar 2026 16:39:09 +0100 Subject: [PATCH 2/2] refactor: deduplicate targets ingestion code in bound load balancer --- hcloud/load_balancers/client.py | 98 +++++++++++------------- hcloud/load_balancers/domain.py | 2 + tests/unit/load_balancers/conftest.py | 2 +- tests/unit/load_balancers/test_client.py | 5 +- 4 files changed, 49 insertions(+), 58 deletions(-) diff --git a/hcloud/load_balancers/client.py b/hcloud/load_balancers/client.py index ca73cd9b..166e86dc 100644 --- a/hcloud/load_balancers/client.py +++ b/hcloud/load_balancers/client.py @@ -93,63 +93,51 @@ def __init__( ] data["private_net"] = private_nets - targets = data.get("targets") - if targets: - tmp_targets = [] - for target in targets: - tmp_target = LoadBalancerTarget(type=target["type"]) - if target["type"] == "server": - tmp_target.server = BoundServer( - client._parent.servers, data=target["server"], complete=False - ) - tmp_target.use_private_ip = target["use_private_ip"] - elif target["type"] == "label_selector": - tmp_target.label_selector = LoadBalancerTargetLabelSelector( - selector=target["label_selector"]["selector"] + def _load_balancer_targets( + raw_targets: list[dict[str, Any]], + ) -> list[LoadBalancerTarget]: + return [_load_balancer_target(raw_target) for raw_target in raw_targets] + + def _load_balancer_target( + raw_target: dict[str, Any], + ) -> LoadBalancerTarget: + result = LoadBalancerTarget(type=raw_target["type"]) + + if raw_target["type"] == "ip": + result.ip = LoadBalancerTargetIP( + ip=raw_target["ip"]["ip"], + ) + + elif raw_target["type"] == "server": + result.server = BoundServer( + client._parent.servers, # pylint: disable=protected-access + data=raw_target["server"], + complete=False, + ) + result.use_private_ip = raw_target["use_private_ip"] + + elif raw_target["type"] == "label_selector": + result.label_selector = LoadBalancerTargetLabelSelector( + selector=raw_target["label_selector"]["selector"] + ) + result.use_private_ip = raw_target["use_private_ip"] + + if (raw_nested_targets := raw_target.get("targets")) is not None: + result.targets = _load_balancer_targets(raw_nested_targets) + + if (raw_health_status := raw_target.get("health_status")) is not None: + result.health_status = [ + LoadBalancerTargetHealthStatus( + listen_port=item["listen_port"], + status=item["status"], ) - tmp_target.use_private_ip = target["use_private_ip"] - nested_targets = target.get("targets", []) - if nested_targets: - tmp_nested = [] - for nested in nested_targets: - nested_target = LoadBalancerTarget(type=nested["type"]) - if nested["type"] == "server": - nested_target.server = BoundServer( - client._parent.servers, - data=nested["server"], - complete=False, - ) - elif nested["type"] == "ip": - nested_target.ip = LoadBalancerTargetIP( - ip=nested["ip"]["ip"] - ) - nested_target.use_private_ip = nested.get("use_private_ip") - nested_health_status = nested.get("health_status") - if nested_health_status is not None: - nested_target.health_status = [ - LoadBalancerTargetHealthStatus( - listen_port=hs["listen_port"], - status=hs["status"], - ) - for hs in nested_health_status - ] - tmp_nested.append(nested_target) - tmp_target.targets = tmp_nested - elif target["type"] == "ip": - tmp_target.ip = LoadBalancerTargetIP(ip=target["ip"]["ip"]) - - target_health_status = target.get("health_status") - if target_health_status is not None: - tmp_target.health_status = [ - LoadBalancerTargetHealthStatus( - listen_port=target_health_status_item["listen_port"], - status=target_health_status_item["status"], - ) - for target_health_status_item in target_health_status - ] + for item in raw_health_status + ] + + return result - tmp_targets.append(tmp_target) - data["targets"] = tmp_targets + if (raw_targets := data.get("targets")) is not None: + data["targets"] = _load_balancer_targets(raw_targets) services = data.get("services") if services: diff --git a/hcloud/load_balancers/domain.py b/hcloud/load_balancers/domain.py index 8ad5d28d..8bad4ded 100644 --- a/hcloud/load_balancers/domain.py +++ b/hcloud/load_balancers/domain.py @@ -411,6 +411,8 @@ class LoadBalancerTarget(BaseDomain): use the private IP instead of primary public IP :param health_status: list List of health statuses of the services on this target. Only present for target types "server" and "ip". + :param targets: list + List of resolved label selector targets. Only present for target types "label_selector". """ __api_properties__ = ( diff --git a/tests/unit/load_balancers/conftest.py b/tests/unit/load_balancers/conftest.py index e505fc1d..8f7f6e6d 100644 --- a/tests/unit/load_balancers/conftest.py +++ b/tests/unit/load_balancers/conftest.py @@ -89,7 +89,7 @@ def response_load_balancer(): }, { "type": "label_selector", - "label_selector": {"selector": "eu"}, + "label_selector": {"selector": "env=prod"}, "use_private_ip": True, "targets": [ { diff --git a/tests/unit/load_balancers/test_client.py b/tests/unit/load_balancers/test_client.py index 20fe6eb7..5945b977 100644 --- a/tests/unit/load_balancers/test_client.py +++ b/tests/unit/load_balancers/test_client.py @@ -19,7 +19,7 @@ ) from hcloud.locations import Location from hcloud.networks import Network -from hcloud.servers import Server +from hcloud.servers import BoundServer, Server from ..conftest import BoundModelTestCase @@ -67,13 +67,14 @@ def test_init_label_selector_nested_targets(self, response_load_balancer): label_selector_target = bound_load_balancer.targets[1] assert label_selector_target.type == "label_selector" - assert label_selector_target.label_selector.selector == "eu" + assert label_selector_target.label_selector.selector == "env=prod" assert label_selector_target.use_private_ip is True assert label_selector_target.targets is not None assert len(label_selector_target.targets) == 1 nested = label_selector_target.targets[0] assert nested.type == "server" + assert isinstance(nested.server, BoundServer) assert nested.server.id == 105054278 assert nested.use_private_ip is True assert nested.health_status is not None