[binary_sensor] Defer integration imports out of module scope

esphome/components/binary_sensor/__init__.py

@@ -3,7 +3,6 @@ from logging import getLogger
 from esphome import automation, core
 from esphome.automation import Condition, maybe_simple_id
 import esphome.codegen as cg
-from esphome.components import mqtt, web_server, zigbee
 from esphome.components.const import CONF_ON_STATE_CHANGE
 import esphome.config_validation as cv
 from esphome.const import (
@@ -61,7 +60,11 @@ from esphome.const import (
 )
 from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import (
+    WEBSERVER_SORTING_SCHEMA,
+    ZIGBEE_BINARY_SENSOR_SCHEMA,
     entity_duplicate_validator,
+    lazy_load_validator,
+    mqtt_component_class,
     queue_entity_register,
     setup_device_class,
     setup_entity,
@@ -433,14 +436,14 @@ def validate_publish_initial_state(value):
 
 
 _BINARY_SENSOR_SCHEMA = (
-    cv.ENTITY_BASE_SCHEMA.extend(web_server.WEBSERVER_SORTING_SCHEMA)
+    cv.ENTITY_BASE_SCHEMA.extend(WEBSERVER_SORTING_SCHEMA)
     .extend(cv.MQTT_COMPONENT_SCHEMA)
-    .extend(zigbee.BINARY_SENSOR_SCHEMA)
+    .extend(ZIGBEE_BINARY_SENSOR_SCHEMA)
     .extend(
         {
             cv.GenerateID(): cv.declare_id(BinarySensor),
             cv.OnlyWith(CONF_MQTT_ID, "mqtt"): cv.declare_id(
-                mqtt.MQTTBinarySensorComponent
+                mqtt_component_class("MQTTBinarySensorComponent")
             ),
             cv.Exclusive(
                 CONF_PUBLISH_INITIAL_STATE, CONF_TRIGGER_ON_INITIAL_STATE
@@ -505,7 +508,7 @@ _BINARY_SENSOR_SCHEMA = (
 
 
 _BINARY_SENSOR_SCHEMA.add_extra(entity_duplicate_validator("binary_sensor"))
-_BINARY_SENSOR_SCHEMA.add_extra(zigbee.validate_binary_sensor)
+_BINARY_SENSOR_SCHEMA.add_extra(lazy_load_validator("zigbee", "validate_binary_sensor"))
 
 
 def binary_sensor_schema(
@@ -608,13 +611,20 @@ async def setup_binary_sensor_core_(var, config):
     CORE.add_job(_build_binary_sensor_automations, var, config)
 
     if mqtt_id := config.get(CONF_MQTT_ID):
+        from esphome.components import mqtt
+
         mqtt_ = cg.new_Pvariable(mqtt_id, var)
         await mqtt.register_mqtt_component(mqtt_, config)
 
     if web_server_config := config.get(CONF_WEB_SERVER):
+        from esphome.components import web_server
+
         await web_server.add_entity_config(var, web_server_config)
 
-    await zigbee.setup_binary_sensor(var, config)
+    if "zigbee" in CORE.loaded_integrations:
+        from esphome.components import zigbee
+
+        await zigbee.setup_binary_sensor(var, config)
 
 
 async def register_binary_sensor(var, config):

esphome/components/web_server/__init__.py

@@ -39,6 +39,15 @@ from esphome.const import (
     PLATFORM_RTL87XX,
 )
 from esphome.core import CORE, CoroPriority, coroutine_with_priority
+
+# Re-exported so entity components and external components keep importing
+# these from web_server; defined in entity_helpers so entity base schemas
+# do not need to import this package.
+from esphome.core.entity_helpers import (  # noqa: F401
+    CONF_SORTING_GROUP_ID,
+    CONF_SORTING_WEIGHT,
+    WEBSERVER_SORTING_SCHEMA,
+)
 import esphome.final_validate as fv
 from esphome.types import ConfigType
 
@@ -49,9 +58,7 @@ AUTO_LOAD = ["json", "web_server_base"]
 AUTH_TYPE_BASIC = "basic"
 AUTH_TYPE_DIGEST = "digest"
 
-CONF_SORTING_GROUP_ID = "sorting_group_id"
 CONF_SORTING_GROUPS = "sorting_groups"
-CONF_SORTING_WEIGHT = "sorting_weight"
 CONF_ALLOWED_ORIGINS = "allowed_origins"
 
 
@@ -225,27 +232,6 @@ sorting_group = {
     cv.Optional(CONF_SORTING_WEIGHT): cv.float_,
 }
 
-WEBSERVER_SORTING_SCHEMA = cv.Schema(
-    {
-        # The per-entity web_server block is cosmetic dashboard ordering —
-        # mark the whole block advanced; the children inherit via the cascade.
-        cv.Optional(CONF_WEB_SERVER, visibility=cv.Visibility.ADVANCED): cv.Schema(
-            {
-                cv.OnlyWith(CONF_WEB_SERVER_ID, "web_server"): cv.use_id(WebServer),
-                cv.Optional(CONF_SORTING_WEIGHT): cv.All(
-                    cv.requires_component("web_server"),
-                    cv.float_,
-                ),
-                cv.Optional(CONF_SORTING_GROUP_ID): cv.All(
-                    cv.requires_component("web_server"),
-                    cv.use_id(cg.int_),
-                ),
-            }
-        )
-    }
-)
-
-
 CONFIG_SCHEMA = cv.All(
     cv.Schema(
         {

esphome/components/zigbee/__init__.py

@@ -15,9 +15,15 @@ from esphome.components.esp32.const import (
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_INTERNAL, CONF_MODEL, CONF_NAME, CONF_ON_START
 from esphome.core import CORE, CoroPriority, coroutine_with_priority
+from esphome.core.entity_helpers import (
+    ZIGBEE_BASE_ENTITY_SCHEMA,
+    ZIGBEE_BINARY_SENSOR_SCHEMA,
+)
 from esphome.types import ConfigType
 
-from .const import (
+# CONF_ENDPOINT, CONF_MAX_EP_NUMBER, CONF_REPORT, CONF_USE_DEVICE_TYPE and
+# REPORT are re-exported for existing consumers of this package's namespace.
+from .const import (  # noqa: F401
     CONF_ENDPOINT,
     CONF_MAX_EP_NUMBER,
     CONF_ON_JOIN,
@@ -45,12 +51,7 @@ from .zigbee_esp32 import (
     validate_sensor_esp32,
     zigbee_require_vfs_select,
 )
-from .zigbee_zephyr import (
-    zephyr_binary_sensor,
-    zephyr_number,
-    zephyr_sensor,
-    zephyr_switch,
-)
+from .zigbee_zephyr import zephyr_number, zephyr_sensor, zephyr_switch
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -59,35 +60,10 @@ CODEOWNERS = ["@luar123", "@tomaszduda23"]
 CONFLICTS_WITH = ["openthread"]
 
 
-def _check_report_deprecation(value: str) -> str:
-    if str(value).lower() in ("coordinator", "enable"):
-        _LOGGER.warning(
-            "Report options 'coordinator' and 'enable' are deprecated and will be removed in a future release. Use 'default' instead."
-        )
-    return value
-
-
-BASE_SCHEMA = cv.Schema(
-    {
-        cv.Optional(CONF_REPORT): cv.All(
-            cv.requires_component("zigbee"),
-            cv.requires_component("esp32"),
-            _check_report_deprecation,
-            cv.enum(REPORT, lower=True),
-        ),
-        cv.Optional(CONF_ENDPOINT): cv.All(
-            cv.requires_component("zigbee"),
-            cv.requires_component("esp32"),
-            cv.int_range(1, CONF_MAX_EP_NUMBER),
-        ),
-        cv.Optional(CONF_USE_DEVICE_TYPE): cv.All(
-            cv.requires_component("zigbee"),
-            cv.requires_component("esp32"),
-            cv.boolean,
-        ),
-    }
-)
-BINARY_SENSOR_SCHEMA = cv.Schema({}).extend(BASE_SCHEMA).extend(zephyr_binary_sensor)
+# Defined in esphome.core.entity_helpers so entity base schemas can reference
+# them without importing this package; re-exported here for existing consumers.
+BASE_SCHEMA = ZIGBEE_BASE_ENTITY_SCHEMA
+BINARY_SENSOR_SCHEMA = ZIGBEE_BINARY_SENSOR_SCHEMA
 SENSOR_SCHEMA = cv.Schema({}).extend(BASE_SCHEMA).extend(zephyr_sensor)
 SWITCH_SCHEMA = cv.Schema({}).extend(zephyr_switch)
 NUMBER_SCHEMA = cv.Schema({}).extend(zephyr_number)

esphome/components/zigbee/const.py

@@ -49,6 +49,17 @@ from esphome.const import (
     UNIT_WATT_HOURS,
 )
 
+# The entity schema keys and report enum live in esphome.core.entity_helpers
+# so entity base schemas can use them without importing this package;
+# re-imported here so zigbee code keeps its existing import paths.
+from esphome.core.entity_helpers import (  # noqa: F401  # pylint: disable=unused-import
+    CONF_ENDPOINT,
+    CONF_REPORT,
+    CONF_USE_DEVICE_TYPE,
+    ZIGBEE_MAX_EP_NUMBER as CONF_MAX_EP_NUMBER,
+    ZIGBEE_REPORT as REPORT,
+)
+
 zigbee_ns = cg.esphome_ns.namespace("zigbee")
 ZigbeeComponent = zigbee_ns.class_("ZigbeeComponent", cg.Component)
 ZigbeeAttribute = zigbee_ns.class_("ZigbeeAttribute", cg.Component)
@@ -56,22 +67,10 @@ BinaryAttrs = zigbee_ns.struct("BinaryAttrs")
 AnalogAttrs = zigbee_ns.struct("AnalogAttrs")
 AnalogAttrsOutput = zigbee_ns.struct("AnalogAttrsOutput")
 
-report = zigbee_ns.enum("ZigbeeReportT")
-REPORT = {
-    "coordinator": report.ZIGBEE_REPORT_COORDINATOR,
-    "enable": report.ZIGBEE_REPORT_ENABLE,
-    "force": report.ZIGBEE_REPORT_FORCE,
-    "default": report.ZIGBEE_REPORT_DEFAULT,
-}
-
-CONF_ENDPOINT = "endpoint"
-CONF_MAX_EP_NUMBER = 239
 CONF_ON_JOIN = "on_join"
 CONF_WIPE_ON_BOOT = "wipe_on_boot"
-CONF_REPORT = "report"
 CONF_ROUTER = "router"
 CONF_POWER_SOURCE = "power_source"
-CONF_USE_DEVICE_TYPE = "use_device_type"
 POWER_SOURCE = {
     "UNKNOWN": 0x00,  # ZB_ZCL_BASIC_POWER_SOURCE_UNKNOWN
     "MAINS_SINGLE_PHASE": 0x01,  # ZB_ZCL_BASIC_POWER_SOURCE_MAINS_SINGLE_PHASE

esphome/components/zigbee/const_zephyr.py

@@ -1,6 +1,11 @@
+# These two schema keys live in esphome.core.entity_helpers so entity base
+# schemas can use them without importing this package.
+from esphome.core.entity_helpers import (  # noqa: F401  # pylint: disable=unused-import
+    CONF_ZIGBEE_BINARY_SENSOR,
+    CONF_ZIGBEE_ID,
+)
+
 CONF_MAX_EP_NUMBER_ZEPHYR = 8
-CONF_ZIGBEE_ID = "zigbee_id"
-CONF_ZIGBEE_BINARY_SENSOR = "zigbee_binary_sensor"
 CONF_ZIGBEE_SENSOR = "zigbee_sensor"
 CONF_ZIGBEE_SWITCH = "zigbee_switch"
 CONF_ZIGBEE_NUMBER = "zigbee_number"

esphome/components/zigbee/zigbee_zephyr.py

@@ -57,15 +57,6 @@ ZigbeeSensor = zigbee_ns.class_("ZigbeeSensor", cg.Component)
 ZigbeeSwitch = zigbee_ns.class_("ZigbeeSwitch", cg.Component)
 ZigbeeNumber = zigbee_ns.class_("ZigbeeNumber", cg.Component)
 
-zephyr_binary_sensor = cv.Schema(
-    {
-        cv.OnlyWith(CONF_ZIGBEE_ID, ["nrf52", "zigbee"]): cv.use_id(ZigbeeComponent),
-        cv.OnlyWith(CONF_ZIGBEE_BINARY_SENSOR, ["nrf52", "zigbee"]): cv.declare_id(
-            ZigbeeBinarySensor
-        ),
-    }
-)
-
 zephyr_sensor = cv.Schema(
     {
         cv.OnlyWith(CONF_ZIGBEE_ID, ["nrf52", "zigbee"]): cv.use_id(ZigbeeComponent),

esphome/core/entity_helpers.py

@@ -1,6 +1,7 @@
 from collections.abc import Callable
 from dataclasses import dataclass, field
 import functools
+import importlib
 import logging
 
 import esphome.codegen as cg
@@ -15,6 +16,8 @@ from esphome.const import (
     CONF_INTERNAL,
     CONF_NAME,
     CONF_UNIT_OF_MEASUREMENT,
+    CONF_WEB_SERVER,
+    CONF_WEB_SERVER_ID,
 )
 from esphome.core import CORE, ID, CoroPriority, coroutine_with_priority
 from esphome.core.config import (
@@ -22,7 +25,7 @@ from esphome.core.config import (
     ICON_MAX_LENGTH,
     UNIT_OF_MEASUREMENT_MAX_LENGTH,
 )
-from esphome.cpp_generator import MockObj, RawStatement, add, get_variable
+from esphome.cpp_generator import MockObj, MockObjClass, RawStatement, add, get_variable
 from esphome.cpp_types import App
 import esphome.final_validate as fv
 from esphome.helpers import (
@@ -635,3 +638,126 @@ def entity_duplicate_validator(platform: str) -> Callable[[ConfigType], ConfigTy
         return config
 
     return validator
+
+
+# ---------------------------------------------------------------------------
+# Cross-integration entity schema fragments
+# ---------------------------------------------------------------------------
+#
+# Entity base schemas offer mqtt/web_server/zigbee options, but importing
+# those packages pulls their full dependency chains (mqtt and zigbee both
+# import the esp32 package) into every entity component import. The
+# fragments below are built from cheap primitives instead: MockObjClass
+# identity is string-based, so the class handles here are interchangeable
+# with the ones the integrations declare, and every key is guarded by
+# ``cv.requires_component``/``cv.OnlyWith``, which consult
+# ``CORE.loaded_integrations`` at validation time without importing. The
+# owning integrations re-export the shared schema names and key strings so
+# each stays defined once.
+
+CONF_SORTING_GROUP_ID = "sorting_group_id"
+CONF_SORTING_WEIGHT = "sorting_weight"
+
+_WebServer = cg.esphome_ns.namespace("web_server").class_(
+    "WebServer", cg.Component, cg.Controller
+)
+
+WEBSERVER_SORTING_SCHEMA = cv.Schema(
+    {
+        # The per-entity web_server block is cosmetic dashboard ordering —
+        # mark the whole block advanced; the children inherit via the cascade.
+        cv.Optional(CONF_WEB_SERVER, visibility=cv.Visibility.ADVANCED): cv.Schema(
+            {
+                cv.OnlyWith(CONF_WEB_SERVER_ID, "web_server"): cv.use_id(_WebServer),
+                cv.Optional(CONF_SORTING_WEIGHT): cv.All(
+                    cv.requires_component("web_server"),
+                    cv.float_,
+                ),
+                cv.Optional(CONF_SORTING_GROUP_ID): cv.All(
+                    cv.requires_component("web_server"),
+                    cv.use_id(cg.int_),
+                ),
+            }
+        )
+    }
+)
+
+_mqtt_ns = cg.esphome_ns.namespace("mqtt")
+_MQTTComponent = _mqtt_ns.class_("MQTTComponent", cg.Component)
+
+
+def mqtt_component_class(name: str) -> MockObjClass:
+    """Handle for a per-entity mqtt::<name> companion class."""
+    return _mqtt_ns.class_(name, _MQTTComponent)
+
+
+CONF_ENDPOINT = "endpoint"
+CONF_REPORT = "report"
+CONF_USE_DEVICE_TYPE = "use_device_type"
+CONF_ZIGBEE_ID = "zigbee_id"
+CONF_ZIGBEE_BINARY_SENSOR = "zigbee_binary_sensor"
+ZIGBEE_MAX_EP_NUMBER = 239
+
+_zigbee_ns = cg.esphome_ns.namespace("zigbee")
+_ZigbeeComponent = _zigbee_ns.class_("ZigbeeComponent", cg.Component)
+_ZigbeeBinarySensor = _zigbee_ns.class_("ZigbeeBinarySensor", cg.Component)
+_zigbee_report = _zigbee_ns.enum("ZigbeeReportT")
+ZIGBEE_REPORT = {
+    "coordinator": _zigbee_report.ZIGBEE_REPORT_COORDINATOR,
+    "enable": _zigbee_report.ZIGBEE_REPORT_ENABLE,
+    "force": _zigbee_report.ZIGBEE_REPORT_FORCE,
+    "default": _zigbee_report.ZIGBEE_REPORT_DEFAULT,
+}
+
+
+def _check_report_deprecation(value: str) -> str:
+    if str(value).lower() in ("coordinator", "enable"):
+        _LOGGER.warning(
+            "Report options 'coordinator' and 'enable' are deprecated and will be removed in a future release. Use 'default' instead."
+        )
+    return value
+
+
+ZIGBEE_BASE_ENTITY_SCHEMA = cv.Schema(
+    {
+        cv.Optional(CONF_REPORT): cv.All(
+            cv.requires_component("zigbee"),
+            cv.requires_component("esp32"),
+            _check_report_deprecation,
+            cv.enum(ZIGBEE_REPORT, lower=True),
+        ),
+        cv.Optional(CONF_ENDPOINT): cv.All(
+            cv.requires_component("zigbee"),
+            cv.requires_component("esp32"),
+            cv.int_range(1, ZIGBEE_MAX_EP_NUMBER),
+        ),
+        cv.Optional(CONF_USE_DEVICE_TYPE): cv.All(
+            cv.requires_component("zigbee"),
+            cv.requires_component("esp32"),
+            cv.boolean,
+        ),
+    }
+)
+
+ZIGBEE_BINARY_SENSOR_SCHEMA = ZIGBEE_BASE_ENTITY_SCHEMA.extend(
+    {
+        cv.OnlyWith(CONF_ZIGBEE_ID, ["nrf52", "zigbee"]): cv.use_id(_ZigbeeComponent),
+        cv.OnlyWith(CONF_ZIGBEE_BINARY_SENSOR, ["nrf52", "zigbee"]): cv.declare_id(
+            _ZigbeeBinarySensor
+        ),
+    }
+)
+
+
+def lazy_load_validator(
+    component: str, name: str
+) -> Callable[[ConfigType], ConfigType]:
+    """Schema extra delegating to ``components.<component>.<name>`` when loaded."""
+
+    def validator(config: ConfigType) -> ConfigType:
+        if component not in CORE.loaded_integrations:
+            return config
+        module = importlib.import_module(f"esphome.components.{component}")
+        return getattr(module, name)(config)
+
+    return validator

tests/unit_tests/test_lazy_imports.py

@@ -190,6 +190,40 @@ def test_api_client_does_not_import_heavy_modules() -> None:
     )
 
 
+def test_binary_sensor_does_not_import_integrations() -> None:
+    """An entity component must not drag in its optional integrations.
+
+    binary_sensor's mqtt/web_server/zigbee schema fragments live in
+    ``esphome.core.entity_helpers``; the integration packages (and the
+    esp32/logger chains mqtt and zigbee pull in) must only load when the
+    user's config actually uses them.
+    """
+    allowed = (
+        "esphome.components",
+        "esphome.components.binary_sensor",
+        "esphome.components.const",
+    )
+    check = (
+        "import sys; import esphome.components.binary_sensor; "
+        f"leaked = [m for m in sys.modules "
+        f"if m.startswith('esphome.components') and m not in {allowed!r}]; "
+        "print(','.join(leaked))"
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", check],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    leaked = result.stdout.strip()
+    assert not leaked, (
+        f"esphome.components.binary_sensor imports integration packages at "
+        f"top level: {leaked}. Keep the shared schema fragments in "
+        "esphome.core.entity_helpers and import the integration inside "
+        "to_code instead."
+    )
+
+
 def test_stacktrace_does_not_import_heavy_modules() -> None:
     """``esphome.stacktrace`` guards its own docstring's contract.