Feat: move to models and add consume_flux

plesna/compute/consume_flux.py

@@ -0,0 +1,8 @@
+from plesna.models.flux import Flux, FluxMetaData
+
+
+def consume_flux(flux: Flux) -> FluxMetaData:
+    metadata = flux.transformation.function(
+        sources=flux.sources, targets=flux.targets, **flux.transformation.extra_kwrds
+    )
+    return FluxMetaData(data=metadata)

plesna/datastore/datacatalogue.py

@@ -0,0 +1,21 @@
+import abc
+
+
+class DataCatalogue:
+    def __init__(self):
+        pass
+
+    @abc.abstractmethod
+    def schemas(self) -> dict[str:str]:
+        """List schemas"""
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def tables(self, schema) -> dict[str:str]:
+        """List table in schema"""
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def infos(self, table: str, schema: str) -> dict[str, str]:
+        """Get infos about the table"""
+        raise NotImplementedError

plesna/datastore/datastore.py

@@ -0,0 +1,3 @@
+class DataStore:
+    def __init__(self, name):
+        self._name

plesna/datastore/fs_datacatalogue.py

@@ -0,0 +1,83 @@
+from pathlib import Path
+
+from pydantic import BaseModel, computed_field
+
+from plesna.models.storage import Schema, Table
+
+from .datacatalogue import DataCatalogue
+
+
+class FSSchema(BaseModel):
+    path: Path
+
+    @computed_field
+    @property
+    def ref(self) -> Schema:
+        return Schema(
+            id=str(self.path),
+            value=str(self.path),
+        )
+
+
+class FSTable(BaseModel):
+    path: Path
+
+    @computed_field
+    @property
+    def ref(self) -> Table:
+        return Table(
+            id=str(self.path),
+            value=str(self.path),
+        )
+
+
+class FSDataCatalogue(DataCatalogue):
+    """DataCatalogue based on files tree structure"""
+
+    def __init__(self, name: str, basepath: str = "."):
+        self._basepath = Path(basepath)
+        self.name = name
+
+        assert self._basepath.exists()
+
+    def ls(
+        self, dir="", only_files=False, only_directories=False, recursive=False
+    ) -> list[str]:
+        dirpath = self._basepath / dir
+
+        if only_files:
+            return [
+                str(f.relative_to(dirpath))
+                for f in dirpath.iterdir()
+                if not f.is_dir() and not str(f).startswith(".")
+            ]
+
+        if only_directories:
+            if recursive:
+                return [
+                    str(f[0].relative_to(dirpath))
+                    for f in dirpath.walk()
+                    if not str(f).startswith(".")
+                ]
+
+            return [
+                str(f.relative_to(dirpath))
+                for f in dirpath.iterdir()
+                if f.is_dir() and not str(f).startswith(".")
+            ]
+
+        return [
+            str(f.relative_to(dirpath))
+            for f in dirpath.iterdir()
+            if not str(f).startswith(".")
+        ]
+
+    def schemas(self) -> dict[str, FSSchema]:
+        """List schemas (sub directories within basepath)"""
+        subdirectories = self.ls("", only_directories=True, recursive=True)
+        return {str(path): FSSchema(path=path) for path in subdirectories}
+
+    def tables(self, schema_id=".") -> dict[str, FSTable]:
+        """List table in schema (which are files in the directory)"""
+        schema_path = schema_id
+        return {path: FSTable(path=path) for path in self.ls(schema_path, only_files=True)}

plesna/graph/graph.py

@@ -0,0 +1,98 @@
+from functools import reduce
+from typing import Callable
+
+from pydantic import BaseModel
+
+
+class Node(BaseModel):
+    name: str
+    infos: dict = {}
+
+    def __hash__(self):
+        return hash(self.name)
+
+
+class Edge(BaseModel):
+    arrow_name: str
+    source: Node
+    target: Node
+    edge_kwrds: dict = {}
+
+
+class Graph:
+    def __init__(self, nodes: list[Node] = [], edges: list[Edge] = []):
+        self._edges = []
+        self._nodes = set()
+        self.add_edges(edges)
+        self.add_nodes(nodes)
+
+    def add_node(self, node: Node):
+        self._nodes.add(node)
+
+    def add_nodes(self, nodes: list[Node]):
+        for node in nodes:
+            self.add_node(node)
+
+    def add_edge(self, edge: Edge):
+        self._edges.append(edge)
+        self.add_node(edge.source)
+        self.add_node(edge.target)
+
+    def add_edges(self, edges: list[Edge]):
+        for edge in edges:
+            self.add_edge(edge)
+
+    @property
+    def nodes(self):
+        return self._nodes
+
+    @property
+    def edges(self):
+        return self._edges
+
+    def get_edges_from(self, node: Node) -> list[Edge]:
+        """Get all edges which have the node as source"""
+        return [edge for edge in self._edges if edge.source == node]
+
+    def get_edges_to(self, node: Node) -> list[Edge]:
+        """Get all edges which have the node as target"""
+        return [edge for edge in self._edges if edge.target == node]
+
+    def get_direct_targets_from(self, node: Node) -> set[Node]:
+        """Get direct nodes that are accessible from the node"""
+        return set(edge.target for edge in self._edges if edge.source == node)
+
+    def get_targets_from(self, node: Node) -> set[Node]:
+        """Get all nodes that are accessible from the node
+
+        If the graph have a loop, the procedure be in an infinite loop!
+
+        """
+        direct_targets = self.get_direct_targets_from(node)
+        undirect_targets = [self.get_targets_from(n) for n in direct_targets]
+        undirect_targets = reduce(lambda x, y: x.union(y), undirect_targets, set())
+
+        return direct_targets.union(undirect_targets)
+
+    def get_direct_sources_from(self, node: Node) -> set[Node]:
+        """Get direct nodes that are targeted the node"""
+        return set(edge.source for edge in self._edges if edge.target == node)
+
+    def get_sources_from(self, node: Node) -> set[Node]:
+        """Get all nodes that are targeted the node"""
+        direct_sources = self.get_direct_sources_from(node)
+        undirect_sources = [self.get_sources_from(n) for n in direct_sources]
+        undirect_sources = reduce(lambda x, y: x.union(y), undirect_sources, set())
+
+        return direct_sources.union(undirect_sources)
+
+    def is_dag(self) -> bool:
+        visited = set()
+        for node in self._nodes:
+            if node not in visited:
+                try:
+                    targets = self.get_targets_from(node)
+                except RecursionError:
+                    return False
+                visited.union(targets)
+        return True

plesna/graph/graph_set.py

@@ -0,0 +1,36 @@
+from typing import Callable
+
+from pydantic import BaseModel
+
+
+class Node(BaseModel):
+    name: str
+    infos: dict = {}
+
+    def __hash__(self):
+        return hash(self.name)
+
+
+class EdgeOnSet(BaseModel):
+    arrow: Callable
+    sources: dict[str, Node]
+    targets: dict[str, Node]
+    edge_kwrds: dict = {}
+
+
+class GraphSet:
+    def __init__(self):
+        self._edges = []
+        self._node_sets = set()
+
+    def append(self, edge: EdgeOnSet):
+        self._edges.append(edge)
+        self._node_sets.add(frozenset(edge.sources.values()))
+        self._node_sets.add(frozenset(edge.targets.values()))
+
+    @property
+    def node_sets(self):
+        return self._node_sets
+
+    def is_valid_dag(self):
+        pass

plesna/models/flux.py

@@ -0,0 +1,14 @@
+from pydantic import BaseModel
+
+from plesna.models.storage import Table
+from plesna.models.transformation import Transformation
+
+
+class Flux(BaseModel):
+    sources: dict[str, Table]
+    targets: dict[str, Table]
+    transformation: Transformation
+
+
+class FluxMetaData(BaseModel):
+    data: dict

plesna/models/storage.py

@@ -0,0 +1,25 @@
+from pydantic import BaseModel
+
+
+class Schema(BaseModel):
+    """Logical agregation for Table
+
+    id: uniq identifier for the schema
+    value: string which describe where to find the schema in the storage system
+
+    """
+
+    id: str
+    value: str
+
+
+class Table(BaseModel):
+    """Place where data are stored
+
+    id: uniq identifier for the table
+    value: string which describe where to find the table in the storage system
+
+    """
+
+    id: str
+    value: str

plesna/models/transformation.py

@@ -0,0 +1,15 @@
+from collections.abc import Callable
+
+from pydantic import BaseModel
+
+
+class Transformation(BaseModel):
+    """
+    The function have to have at least 2 arguments: sources and targets
+    Other arguments will came throught extra_kwrds
+
+    The function will have to return metadata as dict
+    """
+
+    function: Callable
+    extra_kwrds: dict = {}

tests/compute/test_consume_flux.py

@@ -0,0 +1,35 @@
+from plesna.compute.consume_flux import consume_flux
+from plesna.models.flux import Flux
+from plesna.models.storage import Table
+from plesna.models.transformation import Transformation
+
+
+def test_consume_flux():
+    sources = {
+        "src1": Table(id="src1", value="here"),
+        "src2": Table(id="src2", value="here"),
+    }
+    targets = {
+        "tgt1": Table(id="tgt1", value="this"),
+        "tgt2": Table(id="tgt2", value="that"),
+    }
+
+    def func(sources, targets, **kwrds):
+        return {
+            "sources": len(sources),
+            "targets": len(targets),
+            "kwrds": len(kwrds),
+        }
+
+    flux = Flux(
+        sources=sources,
+        targets=targets,
+        transformation=Transformation(function=func, extra_kwrds={"extra": "super"}),
+    )
+
+    meta = consume_flux(flux)
+    assert meta.data == {
+        "sources": 2,
+        "targets": 2,
+        "kwrds": 1,
+    }

tests/datastore/test_fs_datacatalogue.py

@@ -0,0 +1,72 @@
+import shutil
+from pathlib import Path
+
+import pytest
+
+from plesna.datastore.fs_datacatalogue import FSDataCatalogue
+
+FIXTURE_DIR = Path(__file__).parent / Path("./fs_files/")
+
+
+@pytest.fixture
+def location(tmp_path):
+    loc = tmp_path
+    username_loc = loc / "username"
+    username_loc.mkdir()
+    salary_loc = loc / "salary"
+    salary_loc.mkdir()
+    example_src = FIXTURE_DIR
+    assert example_src.exists()
+
+    for f in example_src.glob("*"):
+        if "username" in str(f):
+            shutil.copy(f, username_loc)
+        else:
+            shutil.copy(f, salary_loc)
+
+    return loc
+
+
+def test_init(location):
+    repo = FSDataCatalogue("example", location)
+    assert repo.ls() == [
+        "username",
+        "salary",
+    ]
+
+    assert repo.ls(recursive=True) == [
+        "username",
+        "salary",
+    ]
+
+
+def test_list_schema(location):
+    repo = FSDataCatalogue("example", location)
+    assert {id: s.model_dump()["ref"]["id"] for id, s in repo.schemas().items()} == {
+        ".": ".",
+        "username": "username",
+        "salary": "salary",
+    }
+    assert {id: s.model_dump()["ref"]["value"] for id, s in repo.schemas().items()} == {
+        ".": ".",
+        "username": "username",
+        "salary": "salary",
+    }
+    assert {id: s.model_dump()["path"] for id, s in repo.schemas().items()} == {
+        ".": Path("."),
+        "username": Path("username"),
+        "salary": Path("salary"),
+    }
+
+
+def test_list_tables(location):
+    repo = FSDataCatalogue("example", location)
+    assert repo.tables() == {}
+    assert {id: t.model_dump()["ref"]["value"] for id,t in repo.tables("username").items()} == {
+        "username.csv": "username.csv",
+        "username-password-recovery-code.xlsx": "username-password-recovery-code.xlsx",
+        "username-password-recovery-code.xls": "username-password-recovery-code.xls",
+    }
+    assert {id: t.model_dump()["ref"]["value"] for id,t in repo.tables("salary").items()} == {
+        "salary.pdf": "salary.pdf",
+    }

tests/graphs/test_graph.py

@@ -0,0 +1,107 @@
+import pytest
+
+from plesna.graph.graph import Edge, Graph, Node
+
+
+def test_append_nodess():
+    nodeA = Node(name="A")
+    nodeB = Node(name="B")
+
+    graph = Graph()
+    graph.add_node(nodeA)
+    graph.add_node(nodeB)
+
+    assert graph.nodes == {nodeA, nodeB}
+
+
+def test_append_edges():
+    nodeA = Node(name="A")
+    nodeB = Node(name="B")
+    nodeC = Node(name="C")
+
+    edge1 = Edge(arrow_name="arrow", source=nodeA, target=nodeC)
+    edge2 = Edge(arrow_name="arrow", source=nodeB, target=nodeC)
+
+    graph = Graph()
+    graph.add_edge(edge1)
+    graph.add_edge(edge2)
+
+    assert graph.nodes == {nodeA, nodeB, nodeC}
+
+
+def test_init_edges_nodes():
+    nodeA = Node(name="A")
+    nodeB = Node(name="B")
+    nodeC = Node(name="C")
+
+    edge1 = Edge(arrow_name="arrow", source=nodeB, target=nodeC)
+
+    graph = Graph()
+    graph.add_node(nodeA)
+    graph.add_edge(edge1)
+
+    assert graph.nodes == {nodeA, nodeB, nodeC}
+
+
+@pytest.fixture
+def nodes():
+    return {
+        "A": Node(name="A"),
+        "B": Node(name="B"),
+        "C": Node(name="C"),
+        "D": Node(name="D"),
+    }
+
+
+@pytest.fixture
+def dag_edges(nodes):
+    return {
+        "1": Edge(arrow_name="arrow", source=nodes["A"], target=nodes["C"]),
+        "2": Edge(arrow_name="arrow", source=nodes["B"], target=nodes["C"]),
+        "3": Edge(arrow_name="arrow", source=nodes["C"], target=nodes["D"]),
+    }
+
+
+@pytest.fixture
+def notdag_edges(nodes):
+    return {
+        "1": Edge(arrow_name="arrow", source=nodes["A"], target=nodes["C"]),
+        "2": Edge(arrow_name="arrow", source=nodes["B"], target=nodes["C"]),
+        "3": Edge(arrow_name="arrow", source=nodes["C"], target=nodes["D"]),
+        "4": Edge(arrow_name="arrow", source=nodes["D"], target=nodes["B"]),
+    }
+
+
+def test_get_edges_from(nodes, dag_edges):
+    edges = dag_edges
+    graph = Graph(edges=edges.values())
+    assert graph.get_edges_from(nodes["A"]) == [edges["1"]]
+
+
+def test_get_targets_from(nodes, dag_edges):
+    edges = dag_edges
+    graph = Graph(edges=edges.values())
+    assert graph.get_direct_targets_from(nodes["A"]) == set([nodes["C"]])
+    assert graph.get_direct_targets_from(nodes["C"]) == set([nodes["D"]])
+    assert graph.get_direct_targets_from(nodes["D"]) == set()
+    assert graph.get_targets_from(nodes["A"]) == set([nodes["C"], nodes["D"]])
+
+
+def test_get_sources_from(nodes, dag_edges):
+    edges = dag_edges
+    graph = Graph(edges=edges.values())
+    assert graph.get_direct_sources_from(nodes["A"]) == set()
+    assert graph.get_direct_sources_from(nodes["C"]) == set([nodes["A"], nodes["B"]])
+    assert graph.get_direct_sources_from(nodes["D"]) == set([nodes["C"]])
+
+    assert graph.get_sources_from(nodes["D"]) == set(
+        [nodes["A"], nodes["B"], nodes["C"]]
+    )
+
+
+def test_valid_dage(dag_edges, notdag_edges):
+    graph = Graph(edges=dag_edges.values())
+    assert graph.is_dag()
+
+    graph = Graph(edges=notdag_edges.values())
+    assert not graph.is_dag()

tests/graphs/test_graph_set.py

@@ -0,0 +1,18 @@
+from plesna.graph.graph_set import EdgeOnSet, GraphSet, Node
+
+
+def test_init():
+    nodeA = Node(name="A")
+    nodeB = Node(name="B")
+    nodeC = Node(name="C")
+
+    def arrow(sources, targets):
+        targets["C"].infos["res"] = sources["A"].name + sources["B"].name
+
+    edge1 = EdgeOnSet(
+        arrow=arrow, sources={"A": nodeA, "B": nodeB}, targets={"C": nodeC}
+    )
+    graph_set = GraphSet()
+    graph_set.append(edge1)
+
+    assert graph_set.node_sets == {frozenset([nodeA, nodeB]), frozenset([nodeC])}

tests/repository/test_fs_repository.py

@@ -1,84 +0,0 @@
-import shutil
-from pathlib import Path
-
-import pytest
-from pandas import pandas
-
-from dashboard.libs.repository.fs_repository import FSRepository
-
-EXAMPLE_DIR = "./tests/repository/fs_examples/"
-
-
-@pytest.fixture
-def location(tmp_path):
-    loc = tmp_path
-    username_loc = loc / "username"
-    username_loc.mkdir()
-    salary_loc = loc / "salary"
-    salary_loc.mkdir()
-    example_src = Path(EXAMPLE_DIR)
-
-    for f in example_src.glob("*"):
-        if "username" in str(f):
-            shutil.copy(f, username_loc)
-        else:
-            shutil.copy(f, salary_loc)
-
-    return loc
-
-
-def test_init(location):
-    repo = FSRepository("example", location)
-    assert repo.ls() == [
-        "username",
-        "salary",
-    ]
-    assert repo.schemas() == [
-        ".",
-        "username",
-        "salary",
-    ]
-
-    assert repo.tables() == []
-    assert repo.tables("username") == [
-        "username.csv",
-        "username-password-recovery-code.xlsx",
-        "username-password-recovery-code.xls",
-    ]
-    assert repo.tables("salary") == ["salary.pdf"]
-
-
-def test_read_csv(location):
-    repo = FSRepository("example", location)
-    username = repo.read("username.csv", "username", delimiter=";")
-    assert list(username.columns) == [
-        "Username",
-        "Identifier",
-        "First name",
-        "Last name",
-    ]
-    assert len(username.index) == 5
-
-
-def test_fake_read_xlsx(location):
-    repo = FSRepository("example", location)
-    df = pandas.read_excel(
-        location / "username" / "username-password-recovery-code.xls"
-    )
-    print(df)
-
-
-def test_read_xlsx(location):
-    repo = FSRepository("example", location)
-    username = repo.read("username-password-recovery-code.xls", "username")
-    assert list(username.columns) == [
-        "Username",
-        "Identifier",
-        "One-time password",
-        "Recovery code",
-        "First name",
-        "Last name",
-        "Department",
-        "Location",
-    ]
-    assert len(username.index) == 5