Feat: add execute_flux

plesna/dataplatform.py

@@ -1,5 +1,6 @@
+from plesna.compute.consume_flux import consume_flux
 from plesna.graph.graph_set import GraphSet
-from plesna.models.flux import Flux
+from plesna.models.flux import Flux, FluxMetaData
 from plesna.storage.repository.repository import Repository
 
 
@@ -41,3 +42,8 @@ class DataPlateform:
 
     def flux(self, name: str) -> Flux:
         return self._fluxes[name]
+
+    def execute_flux(self, name: str) -> FluxMetaData:
+        if name not in self._fluxes:
+            raise DataPlateformError("The flux {name} is not registered")
+        return consume_flux(self._fluxes[name])

tests/dataplatform/test_dataplateform.py

@@ -54,21 +54,21 @@ def test_add_repository(
 
 
 @pytest.fixture
-def dataplatform(
-    repository: FSRepository,
-) -> DataPlateform:
-    dp = DataPlateform()
-    dp.add_repository("test", repository)
-    return dp
+def foo_flux(repository: FSRepository) -> Flux:
+    src = {"username": repository.table("raw", "username")}
+    targets = {"username": repository.table("bronze", "username")}
 
+    def foo(sources, targets):
+        return {"who": "foo"}
 
-def test_listing_content(dataplatform: DataPlateform):
-    assert dataplatform.repository("test").schemas() == ["raw", "bronze", "silver"]
-    assert dataplatform.repository("test").schema("raw").tables == [
-        "recovery",
-        "username",
-        "salary",
-    ]
+    extra_kwrds = {}
+
+    flux = Flux(
+        sources=src,
+        targets=targets,
+        transformation=Transformation(function=foo, extra_kwrds=extra_kwrds),
+    )
+    return flux
 
 
 @pytest.fixture
@@ -89,7 +89,10 @@ def copy_flux(repository: FSRepository) -> Flux:
     return flux
 
 
-def test_add_flux(dataplatform: DataPlateform, copy_flux: Flux):
+def test_add_flux(repository: FSRepository, copy_flux: Flux):
+    dataplatform = DataPlateform()
+    dataplatform.add_repository("test", repository)
+
     dataplatform.add_flux(name="copy_flux", flux=copy_flux)
     assert dataplatform.fluxes == ["copy_flux"]
     dataplatform.add_flux(name="copy_flux_bis", flux=copy_flux)
@@ -97,3 +100,32 @@ def test_add_flux(dataplatform: DataPlateform, copy_flux: Flux):
 
     assert dataplatform.flux("copy_flux") == copy_flux
     assert dataplatform.flux("copy_flux_bis") == copy_flux
+
+
+@pytest.fixture
+def dataplatform(
+    repository: FSRepository,
+    foo_flux: Flux,
+    copy_flux: Flux,
+) -> DataPlateform:
+    dp = DataPlateform()
+
+    dp.add_repository("test", repository)
+
+    dp.add_flux("foo", foo_flux)
+    dp.add_flux("copy", copy_flux)
+    return dp
+
+
+def test_listing_content(dataplatform: DataPlateform):
+    assert dataplatform.repository("test").schemas() == ["raw", "bronze", "silver"]
+    assert dataplatform.repository("test").schema("raw").tables == [
+        "recovery",
+        "username",
+        "salary",
+    ]
+
+
+def test_execute_flux(dataplatform: DataPlateform):
+    meta = dataplatform.execute_flux("foo")
+    assert meta.data == {"who": "foo"}