diff --git a/backend/app/flow/nodes/delay.py b/backend/app/flow/nodes/delay.py index daba71e..ba62a2e 100644 --- a/backend/app/flow/nodes/delay.py +++ b/backend/app/flow/nodes/delay.py @@ -106,7 +106,7 @@ class DelayNode(Node): mapping: dict[str, str] = {} cron: str | None = None - def __init__(self, params: dict[str, Any] | None = None, **kwargs): + def __init__(self, params: dict[str, Any] | None = None, **kwargs: Any) -> None: cfg = self.Params.model_validate(params or {}) self.delay = cfg.delay self.interval = cfg.interval @@ -114,8 +114,8 @@ class DelayNode(Node): self.cron_expr = cfg.cron self.ts = 0.0 self.last_input: dict[str, Any] = {} - self._cron_task = None - self._stop_cron = None + self._cron_task: asyncio.Task[None] | None = None + self._stop_cron: asyncio.Event | None = None super().__init__(f=self._f, params=params, **kwargs) @@ -128,7 +128,7 @@ class DelayNode(Node): for i, o in zip(self.input_ports, self.output_ports, strict=False) ] - def _f(self, params, **kwargs): + def _f(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: """Forward messages with optional delay, rate-limiting, and alarm.""" logger.info("[%s] Received %s", self.name, kwargs) @@ -220,12 +220,12 @@ class DelayNode(Node): self._stop_cron = None logger.info("Stopped cron for node '%s'", self.name) - async def _cron_loop(self): + async def _cron_loop(self) -> None: """Background loop that sleeps until the next cron tick and triggers.""" from datetime import datetime try: - from croniter import croniter + from croniter import croniter # type: ignore[import-untyped] except ImportError: logger.error( "croniter package is required for cron scheduling. " @@ -243,7 +243,9 @@ class DelayNode(Node): cron = croniter(self.cron_expr, datetime.now()) - while not (self._stop_cron and self._stop_cron.is_set()): + # Walrus so the body can use the event without re-reading a field + # stop_cron() may have cleared in the meantime. + while (stop := self._stop_cron) is not None and not stop.is_set(): try: # Compute seconds until next tick next_dt = cron.get_next(datetime) @@ -259,7 +261,7 @@ class DelayNode(Node): # Sleep until next tick (wake up on stop signal) try: - await asyncio.wait_for(self._stop_cron.wait(), timeout=wait_seconds) + await asyncio.wait_for(stop.wait(), timeout=wait_seconds) # If we get here, stop was requested break except asyncio.TimeoutError: @@ -283,7 +285,7 @@ class DelayNode(Node): ) raise - async def _trigger_cron(self): + async def _trigger_cron(self) -> None: """Emit data into the pipeline on a cron tick.""" if self._pipeline is None: logger.warning( diff --git a/backend/app/flow/nodes/http.py b/backend/app/flow/nodes/http.py index 1e0d873..67fc688 100644 --- a/backend/app/flow/nodes/http.py +++ b/backend/app/flow/nodes/http.py @@ -204,7 +204,7 @@ class HttpNode(Node): ) @staticmethod - def _noop_trigger(params: dict, **kwargs) -> dict | None: + def _noop_trigger(params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: """ No-op function for trigger mode nodes. @@ -213,7 +213,9 @@ class HttpNode(Node): """ return None - def _sender_handler(self, params: dict, **kwargs) -> dict | None: + def _sender_handler( + self, params: dict[str, Any], **kwargs: Any + ) -> dict[str, Any] | None: """ Send HTTP request with pipeline data (sender mode). @@ -329,6 +331,7 @@ class HttpNode(Node): try: # Parse request data + data: dict[str, Any] if self.method == "GET": data = dict(request.query_params) else: # POST diff --git a/backend/app/flow/nodes/influx.py b/backend/app/flow/nodes/influx.py index c5b6d8a..bed16bd 100644 --- a/backend/app/flow/nodes/influx.py +++ b/backend/app/flow/nodes/influx.py @@ -9,7 +9,7 @@ from typing import Any from pydantic import BaseModel, ConfigDict, Field from app.flow.messages import MessageSpec -from app.flow.nodes.base import Node +from app.flow.nodes.base import Node, NodeResult logger = logging.getLogger(__name__) @@ -200,7 +200,7 @@ class InfluxDbNode(Node): name=name, ) - def _handler(self, params: dict, **kwargs) -> dict | None: + def _handler(self, params: dict[str, Any], **kwargs: Any) -> dict[str, Any] | None: """ Handle incoming data - write to InfluxDB and optionally query. @@ -224,7 +224,7 @@ class InfluxDbNode(Node): return None - def _write_points(self, data: dict) -> None: + def _write_points(self, data: dict[str, Any]) -> None: """ Write data points to InfluxDB using configuration from ``writes``. @@ -327,7 +327,7 @@ class InfluxDbNode(Node): logger.error("InfluxDB write error in node '%s': %s", self.name, e) raise - def _query_data(self) -> dict: + def _query_data(self) -> dict[str, Any]: """ Query data from InfluxDB based on provides configuration. @@ -399,7 +399,7 @@ class InfluxDbNode(Node): self, measurement: str, field: str, - tags: dict, + tags: dict[str, Any], time_range: str, aggregation: str, ) -> str: @@ -452,7 +452,7 @@ class InfluxDbNode(Node): return "\n".join(query_parts) - def _extract_query_result(self, tables, spec: MessageSpec) -> Any: + def _extract_query_result(self, tables: Any, spec: MessageSpec) -> Any: """ Extract a single value from query result tables. @@ -473,7 +473,9 @@ class InfluxDbNode(Node): return None - def inject(self, outputs: dict | None = None) -> dict | None: + def inject( + self, outputs: dict[str, Any] | None = None, durable: bool | None = None + ) -> NodeResult: """ Inject queried data into the pipeline. @@ -495,4 +497,4 @@ class InfluxDbNode(Node): return None outputs = self._query_data() - return self._pipeline.trigger(self, self._to_messages(outputs)) + return self._pipeline.trigger(self, self._to_messages(outputs), durable=durable) diff --git a/backend/app/flow/nodes/mqtt.py b/backend/app/flow/nodes/mqtt.py index ecf2c29..738f776 100644 --- a/backend/app/flow/nodes/mqtt.py +++ b/backend/app/flow/nodes/mqtt.py @@ -190,11 +190,11 @@ class MqttNode(Node): self.keepalive = cfg.keepalive # Runtime state - self._subscription_task: asyncio.Task | None = None + self._subscription_task: asyncio.Task[None] | None = None self._mqtt_client = None self._stop_event: asyncio.Event | None = None - self._publish_queue: asyncio.Queue[dict] | None = None - self._publisher_task: asyncio.Task | None = None + self._publish_queue: asyncio.Queue[dict[str, Any]] | None = None + self._publisher_task: asyncio.Task[None] | None = None self._loop: asyncio.AbstractEventLoop | None = None # Set default name based on mode and topics @@ -228,7 +228,9 @@ class MqttNode(Node): ) @staticmethod - def _noop_subscriber(params: dict, **kwargs) -> dict | None: + def _noop_subscriber( + params: dict[str, Any], **kwargs: Any + ) -> dict[str, Any] | None: """ No-op function for subscriber mode nodes. @@ -245,7 +247,9 @@ class MqttNode(Node): """ return None - def _publisher_handler(self, params: dict, **kwargs) -> dict | None: + def _publisher_handler( + self, params: dict[str, Any], **kwargs: Any + ) -> dict[str, Any] | None: """ Publish pipeline data to MQTT topic (publisher mode). @@ -277,7 +281,9 @@ class MqttNode(Node): ) return None - def _enqueue(self, queue: asyncio.Queue[dict], data: dict) -> None: + def _enqueue( + self, queue: asyncio.Queue[dict[str, Any]], data: dict[str, Any] + ) -> None: """Queue a payload, dropping the oldest when the broker cannot keep up.""" if queue.full(): try: @@ -317,7 +323,7 @@ class MqttNode(Node): self.report_health("down", str(exc)) raise - async def _publish_once(self, data: dict) -> None: + async def _publish_once(self, data: dict[str, Any]) -> None: """Connect, publish, disconnect — the unstarted node's path.""" import aiomqtt @@ -331,7 +337,7 @@ class MqttNode(Node): ) as client: await self._publish_with(client, data) - async def _publish_with(self, client: Any, data: dict) -> None: + async def _publish_with(self, client: Any, data: dict[str, Any]) -> None: """ Publish messages to their mapped MQTT topics. diff --git a/backend/pyproject.toml b/backend/pyproject.toml index af53696..d1914b4 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -46,18 +46,17 @@ build-backend = "hatchling.build" [tool.mypy] strict = true exclude = ["venv", ".venv", "alembic"] +# influxdb_client's Point builder carries no annotations; calling it is not a +# reason to stop checking the caller (app/flow/nodes/influx.py). +untyped_calls_exclude = ["influxdb_client"] -# The integration nodes still carry the prototype's annotations, and croniter and -# influxdb_client ship no stubs. Shrinking one module at a time as each integration -# is revisited; the rest of the package, and everything around it, is checked strictly. +# Every module is checked strictly. influxdb_client is typed but re-exports its +# names implicitly, which strict mode refuses to follow. (croniter, the other +# untyped dependency, is ignored at its two import sites.) [[tool.mypy.overrides]] -module = [ - "app.flow.nodes.delay", - "app.flow.nodes.http", - "app.flow.nodes.influx", - "app.flow.nodes.mqtt", -] -ignore_errors = true +module = ["influxdb_client"] +implicit_reexport = true + [tool.ruff] target-version = "py310"