diff --git a/appdaemon/adapi.py b/appdaemon/adapi.py index 3bb0f26e1..4ed733b71 100644 --- a/appdaemon/adapi.py +++ b/appdaemon/adapi.py @@ -12,7 +12,7 @@ from datetime import timedelta from logging import Logger from pathlib import Path -from typing import TYPE_CHECKING, Any, Literal, TypeVar, overload +from typing import TYPE_CHECKING, Any, Literal, TypeVar, overload, Generic from appdaemon import dependency from appdaemon import exceptions as ade @@ -24,9 +24,17 @@ from appdaemon.models.config.app import AppConfig from appdaemon.parse import resolve_time_str from appdaemon.state import StateCallbackType +from .utils import get_typing_argument -T = TypeVar("T") +if TYPE_CHECKING: + from .models.config.app import AppConfig + from .plugin_management import PluginBase +T = TypeVar("T") +if sys.version_info >= (3, 13): + ModelType = TypeVar("ModelType", bound="AppConfig", default="AppConfig") +else: + ModelType = TypeVar("ModelType", bound="AppConfig") # Check if the module is being imported using the legacy method if __name__ == Path(__file__).name: @@ -40,12 +48,7 @@ ) -if TYPE_CHECKING: - from .models.config.app import AppConfig - from .plugin_management import PluginBase - - -class ADAPI: +class ADAPI(Generic[ModelType]): """AppDaemon API class. This class includes all native API calls to AppDaemon @@ -73,9 +76,20 @@ class ADAPI: namespace: str _plugin: "PluginBase" - def __init__(self, ad: AppDaemon, config_model: "AppConfig"): + def __init__(self, ad: AppDaemon, config_model: ModelType): + self.__app_config_model_class = get_typing_argument(self) or AppConfig self.AD = ad - self.config_model = config_model + # Re-validate/convert incoming AppConfig to the typed config model if specified + try: + if isinstance(config_model, self.__app_config_model_class): + self.config_model = config_model + else: + data = config_model.model_dump(by_alias=True, exclude_unset=True) + self.config_model = self.__app_config_model_class.model_validate(data) + except Exception: + self.err(f"{self.name} configuration does not match the expected type {self.__app_config_model_class.__name__}") + # Let AppManagement wrappers handle logging/state on failure + raise self.dashboard_dir = None if self.AD.http is not None: @@ -85,12 +99,12 @@ def __init__(self, ad: AppDaemon, config_model: "AppConfig"): self.logger = self._logging.get_child(self.name) self.err = self._logging.get_error().getChild(self.name) - if lvl := config_model.log_level: + if lvl := self.config_model.log_level: self.logger.setLevel(lvl) self.err.setLevel(lvl) self.user_logs = {} - if log_name := config_model.log: + if log_name := self.config_model.log: if user_log := self.get_user_log(log_name): self.logger = user_log @@ -151,17 +165,17 @@ def config_dir(self, value: Path) -> None: self.logger.warning("config_dir is read-only and needs to be set before AppDaemon starts") @property - def config_model(self) -> AppConfig: - """The AppConfig model only for this app.""" + def config_model(self) -> ModelType: + """The AppConfig (or specialized) model only for this app.""" return self._config_model @config_model.setter def config_model(self, new_config: Any) -> None: match new_config: - case AppConfig(): + case self.__app_config_model_class(): self._config_model = new_config case _: - self._config_model = AppConfig.model_validate(new_config) + self._config_model = self.__app_config_model_class.model_validate(new_config) self.args = self._config_model.model_dump(by_alias=True, exclude_unset=True) @property diff --git a/appdaemon/plugins/hass/hassapi.py b/appdaemon/plugins/hass/hassapi.py index 6445bb997..2b2c3af7e 100644 --- a/appdaemon/plugins/hass/hassapi.py +++ b/appdaemon/plugins/hass/hassapi.py @@ -1,10 +1,11 @@ import re +import sys from ast import literal_eval from collections.abc import Iterable from copy import deepcopy from datetime import datetime, timedelta from pathlib import Path -from typing import TYPE_CHECKING, Any, Callable, Literal, Type, overload +from typing import TYPE_CHECKING, Any, Callable, Literal, Type, overload, Generic, TypeVar from appdaemon import exceptions as ade from appdaemon import utils @@ -19,6 +20,9 @@ from appdaemon.plugins.hass.notifications import AndroidNotification from appdaemon.services import ServiceCallback +if TYPE_CHECKING: + from appdaemon.models.config import AppConfig + # Check if the module is being imported using the legacy method if __name__ == Path(__file__).name: from appdaemon.logging import Logging @@ -32,11 +36,12 @@ ) -if TYPE_CHECKING: - from ...models.config.app import AppConfig - +if sys.version_info >= (3, 13): + ModelType = TypeVar("ModelType", bound="AppConfig", default="AppConfig") +else: + ModelType = TypeVar("ModelType", bound="AppConfig") -class Hass(ADBase, ADAPI): +class Hass(Generic[ModelType], ADBase, ADAPI[ModelType]): """HASS API class for the users to inherit from. This class provides an interface to the HassPlugin object that connects to Home Assistant. @@ -44,7 +49,7 @@ class Hass(ADBase, ADAPI): _plugin: HassPlugin - def __init__(self, ad: AppDaemon, config_model: "AppConfig"): + def __init__(self, ad: AppDaemon, config_model: ModelType): # Call Super Classes ADBase.__init__(self, ad, config_model) ADAPI.__init__(self, ad, config_model) diff --git a/appdaemon/plugins/mqtt/mqttapi.py b/appdaemon/plugins/mqtt/mqttapi.py index 13096101b..a3f5d49ff 100644 --- a/appdaemon/plugins/mqtt/mqttapi.py +++ b/appdaemon/plugins/mqtt/mqttapi.py @@ -1,5 +1,6 @@ +import sys from pathlib import Path -from typing import TYPE_CHECKING, Any, Callable, Optional, Union +from typing import TYPE_CHECKING, Any, Callable, Optional, Union, TypeVar, Generic import appdaemon.adapi as adapi import appdaemon.adbase as adbase @@ -24,8 +25,12 @@ "To use the Mqtt plugin use 'from appdaemon.plugins import mqtt' instead.", ) +if sys.version_info >= (3, 13): + ModelType = TypeVar("ModelType", bound="AppConfig", default="AppConfig") +else: + ModelType = TypeVar("ModelType", bound="AppConfig") -class Mqtt(adbase.ADBase, adapi.ADAPI): +class Mqtt(Generic[ModelType], adbase.ADBase, adapi.ADAPI[ModelType]): """ A list of API calls and information specific to the MQTT plugin. @@ -73,7 +78,7 @@ def initialize(self): _plugin: "MqttPlugin" - def __init__(self, ad: AppDaemon, config_model: "AppConfig"): + def __init__(self, ad: AppDaemon, config_model: ModelType): # Call Super Classes adbase.ADBase.__init__(self, ad, config_model) adapi.ADAPI.__init__(self, ad, config_model) diff --git a/appdaemon/utils.py b/appdaemon/utils.py index c8c46c67a..39a3ac5c7 100644 --- a/appdaemon/utils.py +++ b/appdaemon/utils.py @@ -23,7 +23,7 @@ from logging import Logger from pathlib import Path from time import perf_counter -from typing import TYPE_CHECKING, Any, Callable, Coroutine, Literal, ParamSpec, Protocol, TypeVar +from typing import TYPE_CHECKING, Any, Callable, Coroutine, Literal, ParamSpec, Protocol, TypeVar, Type import dateutil.parser import tomli @@ -46,6 +46,7 @@ file_log = logger.getChild("file") if TYPE_CHECKING: + from .models.config import AppConfig from .adbase import ADBase from .appdaemon import AppDaemon @@ -1236,3 +1237,43 @@ def recursive_get_files(base: Path, suffix: str, exclude: set[str] | None = None yield item elif item.is_dir() and os.access(item, os.R_OK): yield from recursive_get_files(item, suffix, exclude) + +TA = TypeVar("TA", bound="AppConfig") + +def get_typing_argument(object) -> Type[TA]: + """ + This function is used to extract the typing argument of a generic class or object. + + The purpose is to be able to create an instance of such a type during execution time. + + Args: + object: The object or instance for which the typing argument is to be retrieved. + + Returns: + The typing argument (TA) associated with the given object. The specific typing + argument depends on the generic type definition. + """ + from typing import get_args, get_origin + from .models.config import AppConfig + + oc = getattr(object, "__orig_class__", None) or get_origin(object) + if oc is not None: + for arg in get_args(oc): + if issubclass(arg, AppConfig): + return arg + + # get_original_bases was added in Python 3.12 + if sys.version_info >= (3, 12): + from types import get_original_bases + bases = get_original_bases(object.__class__) + else: + # For Python 3.10 and 3.11, use __orig_bases__ attribute + bases = getattr(object.__class__, "__orig_bases__", ()) + + for base in bases: + if hasattr(base, "__name__") and base.__name__ in {"ADAPI", "ADBase"}: + for arg in get_args(base): + if issubclass(arg, AppConfig): + return arg + + return AppConfig diff --git a/docs/AD_API_REFERENCE.rst b/docs/AD_API_REFERENCE.rst index f55901ae9..752a35d9a 100644 --- a/docs/AD_API_REFERENCE.rst +++ b/docs/AD_API_REFERENCE.rst @@ -54,6 +54,47 @@ for plugins in multiple namespaces. # handle = self.adapi.run_in(...) # handle = self.adapi.run_every(...) +Typed app configuration (Pydantic models) +----------------------------------------- + +App args can be validated and accessed via a typed model by subclassing +``appdaemon.models.config.app.AppConfig`` and typing ``ADAPI`` with it: +``class MyApp(ADAPI[MyConfig]):``. The model instance is available as +``self.config_model``; the untyped dict ``self.args`` remains available for +backward compatibility. + +.. code:: python + + from appdaemon.adapi import ADAPI + from appdaemon.models.config import AppConfig + + class MyConfig(AppConfig, extra="forbid"): + required_int: int + optional_str: str = "Hello" + + class MyApp(ADAPI[MyConfig]): + def initialize(self): + # Typed access + self.log(f"Typed: {self.config_model.required_int}") + # Legacy access + self.log(f"Legacy: {self.args['required_int']}") + +.. code:: yaml + + # apps.yaml + my_app: + module: my_module + class: MyApp + required_int: 42 + +.. note:: + - Validation errors are logged and prevent the app from starting. + - ``extra="forbid"`` rejects unknown keys; omit it if you want to allow + extra args. + +See also the user guide section on app configuration (apps.yaml) for a +full walkthrough. + Entity Class ------------ @@ -261,6 +302,17 @@ Cancels a predefined sequence. The `entity_id` arg with the sequence full-qualif Reference --------- +Configuration +~~~~~~~~~~~~~ + +.. py:attribute:: appdaemon.adapi.ADAPI.config_model + :type: appdaemon.models.config.app.AppConfig + + Typed view of the app’s configuration. When ``ADAPI`` is used with a + generic parameter (e.g., ``ADAPI[MyConfig]``), this attribute is an instance + of that model, providing IDE-friendly, validated access to app args. + ``self.args`` remains available as a plain dict. + Entity API ~~~~~~~~~~ .. automethod:: appdaemon.entity.Entity.add diff --git a/docs/HASS_API_REFERENCE.rst b/docs/HASS_API_REFERENCE.rst index 28e6f543b..e46c1aa7c 100644 --- a/docs/HASS_API_REFERENCE.rst +++ b/docs/HASS_API_REFERENCE.rst @@ -272,6 +272,57 @@ Create apps using the `Hass` API by inheriting from the :py:class:`Hass `__ to learn other inherited helper functions that can be used by Hass applications. +Typed app configuration (Pydantic models) +~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ + +App args can be validated and accessed via a typed model by subclassing +``appdaemon.models.config.app.AppConfig`` and typing the ``Hass`` API with it: +``class MyApp(Hass[MyConfig]):``. The model instance is available as +``self.config_model``; the untyped dict ``self.args`` remains available for +backward compatibility. + +.. code:: python + + from appdaemon.plugins.hass import Hass + from appdaemon.models.config import AppConfig + + + class MyConfig(AppConfig, extra="forbid"): + light: str + brightness: int = 255 + + + class MyApp(Hass[MyConfig]): + def initialize(self) -> None: + # Typed access + self.call_service( + "light/turn_on", + target=self.config_model.light, + brightness=self.config_model.brightness, + ) + # Legacy access + self.call_service( + "light/turn_on", + target=self.args["light"], + brightness=self.args.get("brightness", 255), + ) + +.. code:: yaml + + # apps.yaml + typed_light_app: + module: my_module + class: MyApp + light: light.kitchen + brightness: 200 + +.. note:: + - Validation errors are logged and prevent the app from starting. + - ``extra="forbid"`` rejects unknown keys; omit it if you want to allow extra args. + - See the `AD API Reference `__ for the generic + ``ADAPI`` usage and the ``config_model`` attribute + (`link `__). + Services ~~~~~~~~ diff --git a/docs/HISTORY.md b/docs/HISTORY.md index 70f97209b..f6fb05f8d 100644 --- a/docs/HISTORY.md +++ b/docs/HISTORY.md @@ -8,6 +8,8 @@ - Add request context logging for failed HASS calls - contributed by [ekutner](https://github.com/ekutner) - Reload modified apps on SIGUSR2 - contributed by [chatziko](https://github.com/chatziko) - Using urlib to create endpoints from URLs - contributed by [cebtenzzre](https://github.com/cebtenzzre) +- Support for typed AppConfiguration for apps, using Pydantic models (ADAPI[MyConfig]); + also supported in Hass and MQTT APIs. Contributed by [TCampmany](https://github.com/tcampmany) **Fixes** diff --git a/docs/MQTT_API_REFERENCE.rst b/docs/MQTT_API_REFERENCE.rst index 9223e56da..3df6e4490 100644 --- a/docs/MQTT_API_REFERENCE.rst +++ b/docs/MQTT_API_REFERENCE.rst @@ -12,11 +12,54 @@ To create apps based on just the MQTT API, use some code like the following: .. code:: python + from appdaemon.plugins.mqtt import Mqtt + + + class MyApp(Mqtt): + def initialize(self): + ... # Your initialization code here + +Typed app configuration (Pydantic models) +----------------------------------------- + + App args can be validated and accessed via a typed model by subclassing + ``appdaemon.models.config.app.AppConfig`` and typing the ``Mqtt`` API with it: + ``class MyApp(mqtt.Mqtt[MyConfig]):``. The model instance is available as + ``self.config_model``; the untyped dict ``self.args`` remains available for + backward compatibility. + + .. code:: python + import mqttapi as mqtt + from appdaemon.models.config import AppConfig - class MyApp(mqtt.Mqtt): + class MyConfig(AppConfig, extra="forbid"): + required_topic: str + qos: int = 0 + class MyApp(mqtt.Mqtt[MyConfig]): def initialize(self): + # Typed access + topic = self.config_model.required_topic + self.mqtt_publish(topic, payload="ON", qos=self.config_model.qos) + # Legacy access + self.call_service("mqtt/publish", topic=self.args["required_topic"], payload="ON") + + .. code:: yaml + + # apps.yaml + my_mqtt_app: + module: my_module + class: MyApp + required_topic: "homeassistant/bedroom/light" + qos: 1 + + .. note:: + - Validation errors are logged and prevent the app from starting. + - ``extra="forbid"`` rejects unknown keys; omit it if you want to allow extra args. + - See the `AD API Reference `__ for the generic + ``ADAPI`` usage and the ``config_model`` attribute + (`link `__). Making Calls to MQTT -------------------- diff --git a/tests/conf/apps/typed_hello_world/apps.yaml b/tests/conf/apps/typed_hello_world/apps.yaml new file mode 100644 index 000000000..26a08bfc0 --- /dev/null +++ b/tests/conf/apps/typed_hello_world/apps.yaml @@ -0,0 +1,40 @@ +hello_world_1: + module: hello + class: HelloWorld + +hello_world_2: + module: hello + class: HelloWorld + my_kwarg: "asdf" + +hello_world_3: + module: hello + clas: HelloWorld # wrong attribute name + my_kwarg: "asdf" + +typed_hello_world_1: + module: typed_hello + class: TypedHelloWorld + required_int: 1 + optional_str: "Hi!" + +typed_hello_world_2: + module: typed_hello + class: TypedHelloWorld + required_int: 1 + +typed_hello_world_3: + module: typed_hello + class: TypedHelloWorld + # required_int: 1 # missing + +typed_hello_world_4: + module: typed_hello + class: TypedHelloWorld + required_int: "a" # wrong_type + +typed_hello_world_5: + module: typed_hello + class: TypedHelloWorld + required_int: 1 + typed_hello_world: "does not accept extra fields" # Will fail at load time diff --git a/tests/conf/apps/typed_hello_world/typed_hello.py b/tests/conf/apps/typed_hello_world/typed_hello.py new file mode 100755 index 000000000..18cd608c9 --- /dev/null +++ b/tests/conf/apps/typed_hello_world/typed_hello.py @@ -0,0 +1,23 @@ +from appdaemon.adapi import ADAPI +from appdaemon.models.config import AppConfig + + +class TypedAppConfig(AppConfig, extra="forbid"): + required_int: int + optional_str: str = "Hello" + + +class TypedHelloWorld(ADAPI[TypedAppConfig]): + def initialize(self): + self.log("Hello from TypedHelloWorld") + self.log(f"Config type: {type(self.config_model)}") + self.log("Config values are accessible in both way:") + self.log(f" - Legacy: {self.args['required_int']=}") + self.log(f" - Typed: {self.config_model.required_int=}") + + +class HelloWorld(ADAPI): + def initialize(self): + self.log("Hello from AppDaemon") + self.log("You are now ready to run Apps!") + self.log(f"My kwarg: {self.args.get('my_kwarg', 'not set')}") diff --git a/tests/functional/test_typed.py b/tests/functional/test_typed.py new file mode 100644 index 000000000..26e037f08 --- /dev/null +++ b/tests/functional/test_typed.py @@ -0,0 +1,54 @@ +import logging + +import pytest +from appdaemon.appdaemon import AppDaemon + +logger = logging.getLogger("AppDaemon._test") + + +@pytest.mark.ci +@pytest.mark.functional +@pytest.mark.parametrize( + ("app_name", "app_config_class_name"), + [ + ("hello_world_1", "AppConfig"), + ("hello_world_2", "AppConfig"), + ("typed_hello_world_1", "TypedAppConfig"), + ("typed_hello_world_2", "TypedAppConfig"), + ], +) +@pytest.mark.asyncio(loop_scope="session") +async def test_typed_hello_world(ad: AppDaemon, app_name: str, app_config_class_name: str) -> None: + """Run one of the hello world apps and ensure that the startup text is in the logs.""" + + ad.app_dir = ad.config_dir / "apps/typed_hello_world" + assert ad.app_dir.exists(), "App directory does not exist" + logger.info("Test started") + async with ad.app_management.app_run_context(app_name): + assert (app := ad.app_management.get_app(app_name)) is not None, f"App {app_name} not found." + assert ( + app.config_model.__class__.__name__ == app_config_class_name + ), f"App {app_name} config type name {app.__class__.__name__} differs from {app_config_class_name}" + logger.info("Test completed") + + +@pytest.mark.ci +@pytest.mark.functional +@pytest.mark.parametrize( + "app_name", + [ + "hello_world_3", + "typed_hello_world_3", + "typed_hello_world_4", + "typed_hello_world_5", + ], +) +@pytest.mark.asyncio(loop_scope="session") +async def test_incorrect_typed_hello_world(ad: AppDaemon, app_name: str) -> None: + """Run one of the hello world apps and ensure that the startup text is in the logs.""" + + ad.app_dir = ad.config_dir / "apps/typed_hello_world" + assert ad.app_dir.exists(), "App directory does not exist" + logger.info("Test started") + assert ad.app_management.get_app(app_name) is None, f"App {app_name} was suppose to be not found." + logger.info("Test completed")