sascha/esphome
1
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2025-09-16 18:22:22 +01:00
Code Issues Releases Wiki Activity

dry

Browse Source
This commit is contained in:
J. Nick Koston
2025-09-11 19:43:08 -05:00
parent 0be3387d37
commit 801c15a1e0
1 changed files with 18 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
esphome/helpers.py
View File

@@ -177,6 +177,21 @@ def addr_preference_(res: AddrInfo) -> int:
return 1 return 1
def _add_ip_addresses_to_addrinfo(
addresses: list[str], port: int, res: list[AddrInfo]
) -> None:
"""Helper to add IP addresses to addrinfo results with error handling."""
import socket
for addr in addresses:
try:
res += socket.getaddrinfo(
addr, port, proto=socket.IPPROTO_TCP, flags=socket.AI_NUMERICHOST
)
except OSError:
_LOGGER.debug("Failed to parse IP address '%s'", addr)
def resolve_ip_address( def resolve_ip_address(
host: str | list[str], port: int, address_cache: AddressCache | None = None host: str | list[str], port: int, address_cache: AddressCache | None = None
) -> list[AddrInfo]: ) -> list[AddrInfo]:
@@ -203,13 +218,7 @@ def resolve_ip_address(
# Fast path: if all hosts are already IP addresses # Fast path: if all hosts are already IP addresses
if all(is_ip_address(h) for h in hosts): if all(is_ip_address(h) for h in hosts):
for addr in hosts: _add_ip_addresses_to_addrinfo(hosts, port, res)
try:
res += socket.getaddrinfo(
addr, port, proto=socket.IPPROTO_TCP, flags=socket.AI_NUMERICHOST
)
except OSError:
_LOGGER.debug("Failed to parse IP address '%s'", addr)
# Sort by preference # Sort by preference
res.sort(key=addr_preference_) res.sort(key=addr_preference_)
return res return res
@@ -237,13 +246,7 @@ def resolve_ip_address(
uncached_hosts.append(h) uncached_hosts.append(h)
# Process cached addresses (includes direct IPs and cached lookups) # Process cached addresses (includes direct IPs and cached lookups)
for addr in cached_addresses: _add_ip_addresses_to_addrinfo(cached_addresses, port, res)
try:
res += socket.getaddrinfo(
addr, port, proto=socket.IPPROTO_TCP, flags=socket.AI_NUMERICHOST
)
except OSError:
_LOGGER.debug("Failed to parse IP address '%s'", addr)
# If we have uncached hosts (only non-IP hostnames), resolve them # If we have uncached hosts (only non-IP hostnames), resolve them
if uncached_hosts: if uncached_hosts:
@@ -296,14 +299,7 @@ def sort_ip_addresses(address_list: list[str]) -> list[str]:
# First "resolve" all the IP addresses to getaddrinfo() tuples of the form # First "resolve" all the IP addresses to getaddrinfo() tuples of the form
# (family, type, proto, canonname, sockaddr) # (family, type, proto, canonname, sockaddr)
res: list[AddrInfo] = [] res: list[AddrInfo] = []
for addr in address_list: _add_ip_addresses_to_addrinfo(address_list, 0, res)
# This should always work as these are supposed to be IP addresses
try:
res += socket.getaddrinfo(
addr, 0, proto=socket.IPPROTO_TCP, flags=socket.AI_NUMERICHOST
)
except OSError:
_LOGGER.info("Failed to parse IP address '%s'", addr)
# Now use that information to sort them. # Now use that information to sort them.
res.sort(key=addr_preference_) res.sort(key=addr_preference_)