Improve (r)split performance (#162418)

This commit is contained in:
epenet
2026-02-06 15:10:30 +01:00
committed by GitHub
parent 5ad632c34a
commit bc1c24efb1
18 changed files with 24 additions and 22 deletions

View File

@@ -81,7 +81,7 @@ def get_event_types_by_event_class(event_class: str) -> set[str]:
but if there is only one button then it will be
button without a number postfix.
"""
return EVENT_TYPES_BY_EVENT_CLASS.get(event_class.split("_")[0], set())
return EVENT_TYPES_BY_EVENT_CLASS.get(event_class.split("_", maxsplit=1)[0], set())
async def async_validate_trigger_config(

View File

@@ -142,7 +142,7 @@ def best_matching_language_code(
# Use the assist language if supported
if assist_language in SUPPORTED_LANGUAGE_CODES:
return assist_language
language = assist_language.split("-")[0]
language = assist_language.split("-", maxsplit=1)[0]
# Use the agent language if assist and agent start with the same language part
if agent_language is not None and agent_language.startswith(language):

View File

@@ -31,7 +31,7 @@ UNIQUE_ID = "guardian_{0}"
@callback
def async_get_pin_from_discovery_hostname(hostname: str) -> str:
"""Get the device's 4-digit PIN from its zeroconf-discovered hostname."""
return hostname.split(".")[0].split("-")[1]
return hostname.split(".", maxsplit=1)[0].split("-")[1]
@callback

View File

@@ -55,7 +55,7 @@ async def async_create_fix_flow(
) -> RepairsFlow:
"""Create flow."""
if issue_id.split(".")[0] == "integration_not_found":
if issue_id.split(".", maxsplit=1)[0] == "integration_not_found":
assert data
return IntegrationNotFoundFlow(data)
return ConfirmRepairFlow()

View File

@@ -225,7 +225,7 @@ class KrakenSensor(
self._device_name = create_device_name(tracked_asset_pair)
self._attr_unique_id = "_".join(
[
tracked_asset_pair.split("/")[0],
tracked_asset_pair.split("/", maxsplit=1)[0],
tracked_asset_pair.split("/")[1],
description.key,
]
@@ -291,4 +291,4 @@ class KrakenSensor(
def create_device_name(tracked_asset_pair: str) -> str:
"""Create the device name for a given tracked asset pair."""
return f"{tracked_asset_pair.split('/')[0]} {tracked_asset_pair.split('/')[1]}"
return f"{tracked_asset_pair.split('/', maxsplit=1)[0]} {tracked_asset_pair.split('/')[1]}"

View File

@@ -46,7 +46,7 @@ MAX_SCAN_ATTEMPTS: Final = 16
def short_hostname(hostname: str) -> str:
"""Return the first part of the hostname."""
return hostname.split(".")[0]
return hostname.split(".", maxsplit=1)[0]
def human_readable_name(hostname: str, vendor: str, mac_address: str) -> str:

View File

@@ -112,7 +112,7 @@ def _get_title(id_string: str) -> str:
# Format is S://server/share/folder
# If just S: this will be in the mappings; otherwise use the last folder in path.
title = LIBRARY_TITLES_MAPPING.get(
id_string, urllib.parse.unquote(id_string.split("/")[-1])
id_string, urllib.parse.unquote(id_string.rsplit("/", maxsplit=1)[-1])
)
else:
parts = id_string.split("/")

View File

@@ -106,7 +106,7 @@ class FolderSensor(SensorEntity):
self._state: dict[str, Any] | None = None
self._unsub_timer: CALLBACK_TYPE | None = None
self._short_server_id = server_id.split("-")[0]
self._short_server_id = server_id.split("-", maxsplit=1)[0]
self._attr_name = f"{self._short_server_id} {folder_id} {folder_label}"
self._attr_unique_id = f"{self._short_server_id}-{folder_id}"
self._attr_device_info = DeviceInfo(

View File

@@ -206,7 +206,7 @@ class TeslaFleetWallConnectorEntity(
manufacturer="Tesla",
name="Wall Connector",
via_device=(DOMAIN, str(data.id)),
serial_number=din.split("-")[-1],
serial_number=din.rsplit("-", maxsplit=1)[-1],
model=model,
)

View File

@@ -222,7 +222,7 @@ class TeslemetryWallConnectorEntity(TeslemetryPollingEntity):
configuration_url="https://teslemetry.com/console",
name="Wall Connector",
via_device=(DOMAIN, str(data.id)),
serial_number=din.split("-")[-1],
serial_number=din.rsplit("-", maxsplit=1)[-1],
model=model,
)

View File

@@ -153,7 +153,7 @@ class TessieWallConnectorEntity(TessieBaseEntity):
manufacturer="Tesla",
name="Wall Connector",
via_device=(DOMAIN, str(data.id)),
serial_number=din.split("-")[-1],
serial_number=din.rsplit("-", maxsplit=1)[-1],
)
assert data.live_coordinator
super().__init__(data.live_coordinator, key)

View File

@@ -176,7 +176,7 @@ async def _set_paired_camera(obj: Light | Sensor | Doorlock, camera_id: str) ->
async def _set_doorbell_message(obj: Camera, message: str) -> None:
if message.startswith(DoorbellMessageType.CUSTOM_MESSAGE.value):
message = message.split(":")[-1]
message = message.rsplit(":", maxsplit=1)[-1]
await obj.set_lcd_text(DoorbellMessageType.CUSTOM_MESSAGE, text=message)
elif message == TYPE_EMPTY_VALUE:
await obj.set_lcd_text(None)

View File

@@ -202,7 +202,7 @@ async def remove_privacy_zone(call: ServiceCall) -> None:
@callback
def _async_unique_id_to_mac(unique_id: str) -> str:
"""Extract the MAC address from the registry entry unique id."""
return unique_id.split("_")[0]
return unique_id.split("_", maxsplit=1)[0]
async def set_chime_paired_doorbells(call: ServiceCall) -> None:

View File

@@ -96,9 +96,9 @@ def _get_pairing_schema(input_dict: dict[str, Any] | None = None) -> vol.Schema:
def _host_is_same(host1: str, host2: str) -> bool:
"""Check if host1 and host2 are the same."""
host1 = host1.split(":")[0]
host1 = host1.split(":", maxsplit=1)[0]
host1 = host1 if is_ip_address(host1) else socket.gethostbyname(host1)
host2 = host2.split(":")[0]
host2 = host2.split(":", maxsplit=1)[0]
host2 = host2 if is_ip_address(host2) else socket.gethostbyname(host2)
return host1 == host2

View File

@@ -117,10 +117,10 @@ async def async_create_fix_flow(
) -> RepairsFlow:
"""Create flow."""
if issue_id.split(".")[0] == "device_config_file_changed":
if issue_id.split(".", maxsplit=1)[0] == "device_config_file_changed":
assert data
return DeviceConfigFileChangedFlow(data)
if issue_id.split(".")[0] == "migrate_unique_id":
if issue_id.split(".", maxsplit=1)[0] == "migrate_unique_id":
assert data
return MigrateUniqueIDFlow(data)
return ConfirmRepairFlow()

View File

@@ -90,7 +90,7 @@ def log_triggered_template_error(
elif attribute:
target = f" {CONF_ATTRIBUTES}.{attribute}"
logging.getLogger(f"{__package__}.{entity_id.split('.')[0]}").error(
logging.getLogger(f"{__package__}.{entity_id.split('.', maxsplit=1)[0]}").error(
"Error rendering%s template for %s: %s",
target,
entity_id,
@@ -138,7 +138,9 @@ class ValueTemplate(Template):
).strip()
except jinja2.TemplateError as ex:
message = f"Error parsing value for {entity_id}: {ex} (value: {variables['value']}, template: {self.template})"
logger = logging.getLogger(f"{__package__}.{entity_id.split('.')[0]}")
logger = logging.getLogger(
f"{__package__}.{entity_id.split('.', maxsplit=1)[0]}"
)
logger.debug(message)
return error_value

View File

@@ -91,7 +91,7 @@ async def test_buttons_containers_exceptions(
"""Test that Portainer buttons, but this time when they will do boom for sure."""
await setup_integration(hass, mock_config_entry)
action = client_method.split("_")[0]
action = client_method.split("_", maxsplit=1)[0]
entity_id = f"button.practical_morse_{action}_container"
method_mock = getattr(mock_portainer_client, client_method)

View File

@@ -295,7 +295,7 @@ class SoCoMockFactory:
# Generate a different MAC for the non-default speakers.
# otherwise new devices will not be created.
if ip_address != "192.168.42.2":
last_octet = ip_address.split(".")[-1]
last_octet = ip_address.rsplit(".", maxsplit=1)[-1]
my_speaker_info["mac_address"] = f"00-00-00-00-00-{last_octet.zfill(2)}"
mock_soco.get_speaker_info = Mock(return_value=my_speaker_info)
mock_soco.add_to_queue = Mock(return_value=10)