Add pylint checker for sequential async_add_executor_job calls (#170789)

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

homeassistant/components/backblaze_b2/backup.py

@@ -179,6 +179,7 @@ class BackblazeBackupAgent(BackupAgent):
             uploaded_main_file_info = await self._hass.async_add_executor_job(
                 self._bucket.get_file_info_by_name, filename
             )
+            # pylint: disable-next=home-assistant-sequential-executor-jobs
             await self._hass.async_add_executor_job(uploaded_main_file_info.delete)
         except B2Error:
             _LOGGER.warning(
@@ -386,6 +387,7 @@ class BackblazeBackupAgent(BackupAgent):
         )
 
         await self._hass.async_add_executor_job(file.delete)
+        # pylint: disable-next=home-assistant-sequential-executor-jobs
         await self._hass.async_add_executor_job(metadata_file.delete)
 
         self._invalidate_caches(

homeassistant/components/comfoconnect/fan.py

@@ -97,6 +97,7 @@ class ComfoConnectFan(FanEntity):
         await self.hass.async_add_executor_job(
             self._ccb.comfoconnect.register_sensor, SENSOR_FAN_SPEED_MODE
         )
+        # pylint: disable-next=home-assistant-sequential-executor-jobs
         await self.hass.async_add_executor_job(
             self._ccb.comfoconnect.register_sensor, SENSOR_OPERATING_MODE_BIS
         )

homeassistant/components/ezviz/config_flow.py

@@ -127,11 +127,13 @@ class EzvizConfigFlow(ConfigFlow, domain=DOMAIN):
         await self.hass.async_add_executor_job(ezviz_client.login)
 
         # Secondly try to wake hybernating camera.
+        # pylint: disable-next=home-assistant-sequential-executor-jobs
         await self.hass.async_add_executor_job(
             ezviz_client.get_detection_sensibility, data[ATTR_SERIAL]
         )
 
         # Thirdly attempts an authenticated RTSP DESCRIBE request.
+        # pylint: disable-next=home-assistant-sequential-executor-jobs
         await self.hass.async_add_executor_job(_test_camera_rtsp_creds, data)
 
         return self.async_create_entry(

homeassistant/components/fireservicerota/coordinator.py

@@ -179,6 +179,7 @@ class FireServiceRotaClient:
                 self.token_refresh_failure = False
                 await self._hass.async_add_executor_job(self.websocket.start_listener)
 
+                # pylint: disable-next=home-assistant-sequential-executor-jobs
                 return await self._hass.async_add_executor_job(func, *args)
 
     async def async_update(self) -> dict | None:

homeassistant/components/huawei_lte/config_flow.py

@@ -246,6 +246,7 @@ class HuaweiLteConfigFlow(ConfigFlow, domain=DOMAIN):
         info, wlan_settings = await self.hass.async_add_executor_job(
             get_device_info, conn
         )
+        # pylint: disable-next=home-assistant-sequential-executor-jobs
         await self.hass.async_add_executor_job(self._disconnect, conn)
 
         user_input.update(

homeassistant/components/nuki/__init__.py

@@ -162,6 +162,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: NukiConfigEntry) -> bool
             DEFAULT_TIMEOUT,
         )
 
+        # pylint: disable-next=home-assistant-sequential-executor-jobs
         locks, openers = await hass.async_add_executor_job(_get_bridge_devices, bridge)
     except InvalidCredentialsException as err:
         raise exceptions.ConfigEntryAuthFailed from err

homeassistant/components/nuki/config_flow.py

@@ -51,6 +51,7 @@ async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str,
             DEFAULT_TIMEOUT,
         )
 
+        # pylint: disable-next=home-assistant-sequential-executor-jobs
         info = await hass.async_add_executor_job(bridge.info)
     except InvalidCredentialsException as err:
         raise InvalidAuth from err

homeassistant/components/roomba/vacuum.py

@@ -308,6 +308,7 @@ class RoombaVacuumCarpetBoost(RoombaVacuum):
         await self.hass.async_add_executor_job(
             self.vacuum.set_preference, "carpetBoost", str(carpet_boost)
         )
+        # pylint: disable-next=home-assistant-sequential-executor-jobs
         await self.hass.async_add_executor_job(
             self.vacuum.set_preference, "vacHigh", str(high_perf)
         )
@@ -389,6 +390,7 @@ class BraavaJet(IRobotVacuum):
         await self.hass.async_add_executor_job(
             self.vacuum.set_preference, "rankOverlap", overlap
         )
+        # pylint: disable-next=home-assistant-sequential-executor-jobs
         await self.hass.async_add_executor_job(
             self.vacuum.set_preference,
             "padWetness",

homeassistant/components/smappee/config_flow.py

@@ -183,6 +183,7 @@ class SmappeeFlowHandler(
             serial_number = await self.hass.async_add_executor_job(
                 smappee_mqtt.start_and_wait_for_config
             )
+            # pylint: disable-next=home-assistant-sequential-executor-jobs
             await self.hass.async_add_executor_job(smappee_mqtt.stop)
             if serial_number is None:
                 return self.async_abort(reason="cannot_connect")

homeassistant/components/soma/__init__.py

@@ -60,6 +60,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
 async def async_setup_entry(hass: HomeAssistant, entry: SomaConfigEntry) -> bool:
     """Set up Soma from a config entry."""
     api = await hass.async_add_executor_job(SomaApi, entry.data[HOST], entry.data[PORT])
+    # pylint: disable-next=home-assistant-sequential-executor-jobs
     devices = await hass.async_add_executor_job(api.list_devices)
     entry.runtime_data = SomaData(api, devices["shades"])
 

homeassistant/components/starline/config_flow.py

@@ -192,6 +192,7 @@ class StarlineFlowHandler(ConfigFlow, domain=DOMAIN):
             self._app_code = await self.hass.async_add_executor_job(
                 self._auth.get_app_code, self._app_id, self._app_secret
             )
+            # pylint: disable-next=home-assistant-sequential-executor-jobs
             self._app_token = await self.hass.async_add_executor_job(
                 self._auth.get_app_token, self._app_id, self._app_secret, self._app_code
             )

homeassistant/components/tado/coordinator.py

@@ -284,6 +284,7 @@ class TadoDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
 
         try:
             weather = await self.hass.async_add_executor_job(self._tado.get_weather)
+            # pylint: disable-next=home-assistant-sequential-executor-jobs
             geofence = await self.hass.async_add_executor_job(self._tado.get_home_state)
         except RequestException as err:
             _LOGGER.error("Error updating Tado home: %s", err)

homeassistant/components/tami4/config_flow.py

@@ -69,6 +69,7 @@ class Tami4ConfigFlow(ConfigFlow, domain=DOMAIN):
                 refresh_token = await self.hass.async_add_executor_job(
                     Tami4EdgeAPI.submit_otp, self.phone, otp
                 )
+                # pylint: disable-next=home-assistant-sequential-executor-jobs
                 api = await self.hass.async_add_executor_job(
                     Tami4EdgeAPI, refresh_token
                 )

homeassistant/components/vera/__init__.py

@@ -64,6 +64,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: VeraConfigEntry) -> bool
     try:
         all_devices = await hass.async_add_executor_job(controller.get_devices)
 
+        # pylint: disable-next=home-assistant-sequential-executor-jobs
         all_scenes = await hass.async_add_executor_job(controller.get_scenes)
     except RequestException as exception:
         # There was a network related error connecting to the Vera controller.

homeassistant/components/verisure/config_flow.py

@@ -253,6 +253,7 @@ class VerisureConfigFlowHandler(ConfigFlow, domain=DOMAIN):
                 await self.hass.async_add_executor_job(
                     self.verisure.validate_mfa, user_input[CONF_CODE]
                 )
+                # pylint: disable-next=home-assistant-sequential-executor-jobs
                 await self.hass.async_add_executor_job(self.verisure.login)
             except VerisureLoginError as ex:
                 LOGGER.debug("Could not log in to Verisure, %s", ex)

homeassistant/components/yale_smart_alarm/coordinator.py

@@ -44,6 +44,7 @@ class YaleDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
                 self.config_entry.data[CONF_USERNAME],
                 self.config_entry.data[CONF_PASSWORD],
             )
+            # pylint: disable-next=home-assistant-sequential-executor-jobs
             self.locks = await self.hass.async_add_executor_job(self.yale.get_locks)
         except AuthenticationError as error:
             raise ConfigEntryAuthFailed from error

pylint/plugins/pylint_home_assistant/checkers/sequential_executor_jobs.py

@@ -0,0 +1,119 @@
+"""Checker for sequential async_add_executor_job calls.
+
+Multiple sequential ``await hass.async_add_executor_job()`` calls should
+be grouped into a single executor job to avoid unnecessary context switches
+back to the event loop between blocking calls.
+
+https://developers.home-assistant.io/docs/asyncio_working_with_async/#calling-sync-functions-from-async
+"""
+
+from astroid import nodes
+from pylint.checkers import BaseChecker
+from pylint.lint import PyLinter
+
+from pylint_home_assistant.helpers.module_info import is_integration_module
+
+
+def _is_executor_job_await(node: nodes.NodeNG) -> bool:
+    """Return True if *node* is ``await *.async_add_executor_job(...)``."""
+    if not isinstance(node, (nodes.Assign, nodes.AnnAssign, nodes.Expr, nodes.Return)):
+        return False
+
+    value = node.value
+    if value is None:
+        return False
+
+    # Must be an Await
+    if not isinstance(value, nodes.Await):
+        return False
+
+    call = value.value
+    if not isinstance(call, nodes.Call):
+        return False
+
+    return (
+        isinstance(call.func, nodes.Attribute)
+        and call.func.attrname == "async_add_executor_job"
+    )
+
+
+class SequentialExecutorJobsChecker(BaseChecker):
+    """Checker for sequential async_add_executor_job calls."""
+
+    name = "home_assistant_sequential_executor_jobs"
+    priority = -1
+    msgs = {
+        "W7415": (
+            "Sequential `async_add_executor_job` calls should be grouped "
+            "into a single executor job",
+            "home-assistant-sequential-executor-jobs",
+            "Used when multiple await hass.async_add_executor_job() calls "
+            "appear in sequence. Group the blocking operations into a "
+            "single function and call async_add_executor_job once.",
+        ),
+    }
+    options = ()
+
+    _in_integration: bool
+
+    def visit_module(self, node: nodes.Module) -> None:
+        """Track whether we are in an integration module."""
+        self._in_integration = is_integration_module(node.name)
+
+    def visit_functiondef(self, node: nodes.FunctionDef) -> None:
+        """Check for sequential executor job calls."""
+        if not self._in_integration:
+            return
+
+        self._check_body(node.body)
+
+    visit_asyncfunctiondef = visit_functiondef
+
+    def _check_body(self, body: list[nodes.NodeNG]) -> None:
+        """Check a list of statements for sequential executor job calls."""
+        prev_was_executor = False
+
+        for stmt in body:
+            if _is_executor_job_await(stmt):
+                if prev_was_executor:
+                    self.add_message(
+                        "home-assistant-sequential-executor-jobs",
+                        node=stmt,
+                    )
+                prev_was_executor = True
+            else:
+                prev_was_executor = False
+
+                # Recurse into control flow blocks (but not nested functions)
+                if isinstance(stmt, nodes.If):
+                    self._check_body(stmt.body)
+                    self._check_body(stmt.orelse)
+                elif isinstance(stmt, nodes.Try):
+                    self._check_body(stmt.body)
+                    for handler in stmt.handlers:
+                        self._check_body(handler.body)
+                    self._check_body(stmt.orelse)
+                    self._check_body(stmt.finalbody)
+                elif isinstance(
+                    stmt,
+                    (
+                        nodes.With,
+                        nodes.AsyncWith,
+                    ),
+                ):
+                    self._check_body(stmt.body)
+                elif isinstance(
+                    stmt,
+                    (
+                        nodes.For,
+                        nodes.AsyncFor,
+                        nodes.While,
+                    ),
+                ):
+                    self._check_body(stmt.body)
+                    self._check_body(stmt.orelse)
+
+
+def register(linter: PyLinter) -> None:
+    """Register the checker."""
+    linter.register_checker(SequentialExecutorJobsChecker(linter))

tests/pylint/test_sequential_executor_jobs.py

@@ -0,0 +1,269 @@
+"""Tests for the sequential executor jobs checker."""
+
+import astroid
+from pylint.testutils import UnittestLinter
+from pylint.utils.ast_walker import ASTWalker
+from pylint_home_assistant.checkers.sequential_executor_jobs import (
+    SequentialExecutorJobsChecker,
+)
+import pytest
+
+from . import assert_no_messages
+
+
+@pytest.fixture(name="executor_checker")
+def executor_checker_fixture(
+    linter: UnittestLinter,
+) -> SequentialExecutorJobsChecker:
+    """Fixture to provide a sequential executor jobs checker."""
+    return SequentialExecutorJobsChecker(linter)
+
+
+@pytest.mark.parametrize(
+    "code",
+    [
+        pytest.param(
+            """
+async def async_setup(hass, config):
+    await hass.async_add_executor_job(blocking_call)
+""",
+            id="single_call",
+        ),
+        pytest.param(
+            """
+async def async_setup(hass, config):
+    await hass.async_add_executor_job(call_a)
+    result = do_something()
+    await hass.async_add_executor_job(call_b)
+""",
+            id="separated_by_other_statement",
+        ),
+        pytest.param(
+            """
+async def async_setup(hass, config):
+    await hass.async_add_executor_job(call_a)
+    await hass.async_do_something_else()
+    await hass.async_add_executor_job(call_b)
+""",
+            id="separated_by_other_await",
+        ),
+    ],
+)
+def test_no_warning(
+    linter: UnittestLinter,
+    executor_checker: SequentialExecutorJobsChecker,
+    code: str,
+) -> None:
+    """Test cases that should not trigger a warning."""
+    root_node = astroid.parse(code, "homeassistant.components.test_integration")
+    walker = ASTWalker(linter)
+    walker.add_checker(executor_checker)
+
+    with assert_no_messages(linter):
+        walker.walk(root_node)
+
+
+def test_two_sequential_flagged(
+    linter: UnittestLinter,
+    executor_checker: SequentialExecutorJobsChecker,
+) -> None:
+    """Test that two sequential executor jobs are flagged."""
+    root_node = astroid.parse(
+        """
+async def async_setup(hass, config):
+    await hass.async_add_executor_job(call_a)
+    await hass.async_add_executor_job(call_b)
+""",
+        "homeassistant.components.test_integration",
+    )
+    walker = ASTWalker(linter)
+    walker.add_checker(executor_checker)
+    walker.walk(root_node)
+
+    messages = linter.release_messages()
+    assert len(messages) == 1
+    assert messages[0].msg_id == "home-assistant-sequential-executor-jobs"
+
+
+def test_three_sequential_flagged(
+    linter: UnittestLinter,
+    executor_checker: SequentialExecutorJobsChecker,
+) -> None:
+    """Test that three sequential executor jobs flag the second and third."""
+    root_node = astroid.parse(
+        """
+async def async_setup(hass, config):
+    await hass.async_add_executor_job(call_a)
+    await hass.async_add_executor_job(call_b)
+    await hass.async_add_executor_job(call_c)
+""",
+        "homeassistant.components.test_integration",
+    )
+    walker = ASTWalker(linter)
+    walker.add_checker(executor_checker)
+    walker.walk(root_node)
+
+    messages = linter.release_messages()
+    assert len(messages) == 2
+
+
+def test_assigned_sequential_flagged(
+    linter: UnittestLinter,
+    executor_checker: SequentialExecutorJobsChecker,
+) -> None:
+    """Test that assigned sequential executor jobs are also flagged."""
+    root_node = astroid.parse(
+        """
+async def async_setup(hass, config):
+    result_a = await hass.async_add_executor_job(call_a)
+    result_b = await hass.async_add_executor_job(call_b)
+""",
+        "homeassistant.components.test_integration",
+    )
+    walker = ASTWalker(linter)
+    walker.add_checker(executor_checker)
+    walker.walk(root_node)
+
+    messages = linter.release_messages()
+    assert len(messages) == 1
+
+
+def test_inside_try_block_flagged(
+    linter: UnittestLinter,
+    executor_checker: SequentialExecutorJobsChecker,
+) -> None:
+    """Test that sequential calls inside a try block are flagged."""
+    root_node = astroid.parse(
+        """
+async def async_setup(hass, config):
+    try:
+        await hass.async_add_executor_job(call_a)
+        await hass.async_add_executor_job(call_b)
+    except Exception:
+        pass
+""",
+        "homeassistant.components.test_integration",
+    )
+    walker = ASTWalker(linter)
+    walker.add_checker(executor_checker)
+    walker.walk(root_node)
+
+    messages = linter.release_messages()
+    assert len(messages) == 1
+
+
+def test_inside_except_block_flagged(
+    linter: UnittestLinter,
+    executor_checker: SequentialExecutorJobsChecker,
+) -> None:
+    """Test that sequential calls inside an except block are flagged."""
+    root_node = astroid.parse(
+        """
+async def async_setup(hass, config):
+    try:
+        pass
+    except Exception:
+        await hass.async_add_executor_job(call_a)
+        await hass.async_add_executor_job(call_b)
+""",
+        "homeassistant.components.test_integration",
+    )
+    walker = ASTWalker(linter)
+    walker.add_checker(executor_checker)
+    walker.walk(root_node)
+
+    messages = linter.release_messages()
+    assert len(messages) == 1
+    assert messages[0].msg_id == "home-assistant-sequential-executor-jobs"
+
+
+def test_inside_finally_block_flagged(
+    linter: UnittestLinter,
+    executor_checker: SequentialExecutorJobsChecker,
+) -> None:
+    """Test that sequential calls inside a finally block are flagged."""
+    root_node = astroid.parse(
+        """
+async def async_setup(hass, config):
+    try:
+        pass
+    finally:
+        await hass.async_add_executor_job(call_a)
+        await hass.async_add_executor_job(call_b)
+""",
+        "homeassistant.components.test_integration",
+    )
+    walker = ASTWalker(linter)
+    walker.add_checker(executor_checker)
+    walker.walk(root_node)
+
+    messages = linter.release_messages()
+    assert len(messages) == 1
+    assert messages[0].msg_id == "home-assistant-sequential-executor-jobs"
+
+
+def test_inside_for_else_block_flagged(
+    linter: UnittestLinter,
+    executor_checker: SequentialExecutorJobsChecker,
+) -> None:
+    """Test that sequential calls inside a for/else block are flagged."""
+    root_node = astroid.parse(
+        """
+async def async_setup(hass, config):
+    for item in items:
+        pass
+    else:
+        await hass.async_add_executor_job(call_a)
+        await hass.async_add_executor_job(call_b)
+""",
+        "homeassistant.components.test_integration",
+    )
+    walker = ASTWalker(linter)
+    walker.add_checker(executor_checker)
+    walker.walk(root_node)
+
+    messages = linter.release_messages()
+    assert len(messages) == 1
+    assert messages[0].msg_id == "home-assistant-sequential-executor-jobs"
+
+
+def test_return_await_sequential_flagged(
+    linter: UnittestLinter,
+    executor_checker: SequentialExecutorJobsChecker,
+) -> None:
+    """Test that return await following an executor job is flagged."""
+    root_node = astroid.parse(
+        """
+async def async_setup(hass, config):
+    await hass.async_add_executor_job(call_a)
+    return await hass.async_add_executor_job(call_b)
+""",
+        "homeassistant.components.test_integration",
+    )
+    walker = ASTWalker(linter)
+    walker.add_checker(executor_checker)
+    walker.walk(root_node)
+
+    messages = linter.release_messages()
+    assert len(messages) == 1
+    assert messages[0].msg_id == "home-assistant-sequential-executor-jobs"
+
+
+def test_not_integration_module_ignored(
+    linter: UnittestLinter,
+    executor_checker: SequentialExecutorJobsChecker,
+) -> None:
+    """Test that non-integration modules are ignored."""
+    root_node = astroid.parse(
+        """
+async def async_setup(hass, config):
+    await hass.async_add_executor_job(call_a)
+    await hass.async_add_executor_job(call_b)
+""",
+        "tests.components.test_integration",
+    )
+    walker = ASTWalker(linter)
+    walker.add_checker(executor_checker)
+
+    with assert_no_messages(linter):
+        walker.walk(root_node)