Update MQTT configuration and refactor related classes for improved clarity and functionality

Author: legout

examples/hello-world/conf/project.yml

@@ -14,6 +14,11 @@ worker:
     uri: sqlite+aiosqlite:///examples/hello-world/flowerpower.db
     #uri: postgresql+asyncpg://postgres:password@localhost/flowerpower
   event_broker:
-    type: redis
-    uri: redis://localhost:6379
+    #type: redis
+    #uri: redis://localhost:6379
+    type: mqtt
+    host: localhost
+    port: 1883
+    username: null
+    password: null
   max_concurrent_jobs: 100

pyproject.toml

@@ -4,7 +4,7 @@ description = "A simple workflow framework. Hamilton + APScheduler = FlowerPower
 authors = [{ name = "Volker L.", email = "[email protected]" }]
 readme = "README.md"
 requires-python = ">= 3.11"
-version = "0.9.8.6"
+version = "0.9.8.7"
 keywords = ["hamilton", "workflow", "pipeline", "scheduler", "apscheduler", "dask", "ray"]
 dependencies = [
   'aiobotocore<2.18.0',

src/flowerpower/fs/ext.py

@@ -868,7 +868,7 @@ def pydala_dataset(
     path: str,
     partitioning: str | list[str] | pds.Partitioning = None,
     **kwargs,
-) -> ParquetDataset:
+) -> ParquetDataset:  # type: ignore
     """
     Create a pydala dataset.
 

src/flowerpower/io/base.py

@@ -678,7 +678,7 @@ def to_pyarrow_table(self, **kwargs) -> pa.Table:
             self.to_pyarrow_dataset(**kwargs)
         return self._dataset.to_table()
 
-    def to_pydala_dataset(self, **kwargs) -> ParquetDataset:
+    def to_pydala_dataset(self, **kwargs) -> "ParquetDataset":
         if ParquetDataset is None:
             raise ImportError("pydala is not installed.")
         if not hasattr(self, "_pydala_dataset"):

src/flowerpower/io/loader/mqtt.py

@@ -34,20 +34,20 @@ def _to_polars_dataframe(self) -> pl.DataFrame:
     def _to_polars_lazyframe(self) -> pl.LazyFrame:
         return pl.LazyFrame(self.payload)
 
-    def to_poars(self, lazy: bool = False) -> pl.DataFrame | pl.LazyFrame:
+    def to_polars(self, lazy: bool = False) -> pl.DataFrame | pl.LazyFrame:
         if lazy:
             return self._to_polars_lazyframe()
         else:
             return self._to_polars_dataframe()
 
-    def to_duckdb(self, conn: duckdb.DuckDBPyConnection | None = None):
+    def to_duckdb_relation(self, conn: duckdb.DuckDBPyConnection | None = None):
         if self.conn is None:
             if conn is None:
                 conn = duckdb.connect()
             self.conn = conn
         return self.conn.from_arrow(self.to_pyarrow_table())
 
-    def to_dataset(self, **kwargs) -> pds.Dataset:
+    def to_pyarrow_dataset(self, **kwargs) -> pds.Dataset:
         return pds.dataset(self.to_pyarrow_table(), **kwargs)
 
     def register_in_duckdb(

src/flowerpower/mqtt.py

@@ -16,8 +16,8 @@
 class MQTTClient:
     def __init__(
         self,
-        user: str | None = None,
-        pw: str | None = None,
+        username: str | None = None,
+        password: str | None = None,
         host: str | None = "localhost",
         port: int | None = 1883,
         topic: str | None = None,
@@ -26,11 +26,17 @@ def __init__(
         reconnect_rate: int = 2,
         max_reconnect_delay: int = 60,
         transport: str = "tcp",
+        **kwargs,
     ):
+        if "user" in kwargs:
+            username = kwargs["user"]
+        if "pw" in kwargs:
+            password = kwargs["pw"]
+
         self.topic = topic
 
-        self._user = user
-        self._pw = pw
+        self._username = username
+        self._password = password
         self._host = host
         self._port = port
         self._first_reconnect_delay = first_reconnect_delay
@@ -136,8 +142,8 @@ def connect(self) -> Client:
             client_id=f"flowerpower-{random.randint(0, 10000)}",
             transport=self._transport,
             userdata=Munch(
-                user=self._user,
-                pw=self._pw,
+                user=self._username,
+                pw=self._password,
                 host=self._host,
                 port=self._port,
                 topic=self.topic,
@@ -148,8 +154,8 @@ def connect(self) -> Client:
                 transport=self._transport,
             ),
         )
-        if self._pw != "" and self._user != "":
-            client.username_pw_set(self._user, self._pw)
+        if self._password != "" and self._username != "":
+            client.username_pw_set(self._username, self._password)
 
         client.on_connect = self._on_connect
         client.on_disconnect = self._on_disconnect