Feat: add is_dag to Graph

dashboard/app.py

@@ -1,64 +0,0 @@
-import dash
-from dash import Dash, dcc, html
-
-from .datalake import stages
-from .pages import config, home, repository, schema, table
-
-external_scripts = [{"src": "https://cdn.tailwindcss.com"}]
-# external_script = ["https://tailwindcss.com/", {"src": "https://cdn.tailwindcss.com"}]
-
-app = Dash(
-    __name__,
-    use_pages=True,
-    external_scripts=external_scripts,
-    suppress_callback_exceptions=True,
-)
-app.scripts.config.serve_locally = True
-dash.register_page(
-    home.__name__,
-    path="/",
-    layout=home.layout,
-)
-dash.register_page(config.__name__, path="/config", layout=config.layout)
-dash.register_page(
-    repository.__name__,
-    path_template="/repository/<repository_name>",
-    layout=repository.layout_factory(stages),
-)
-dash.register_page(
-    schema.__name__,
-    path_template="/stg/<repository_name>/schema/<schema_name>",
-    layout=schema.layout_factory(stages),
-)
-dash.register_page(
-    table.__name__,
-    path_template="/stg/<repository_name>/schm/<schema_name>/table/<table_name>",
-    layout=table.layout_factory(stages),
-)
-table.callback_factory(app)
-
-app.layout = html.Div(
-    [
-        html.Div(
-            [
-                dcc.Link(
-                    html.H1(
-                        "Plesna",
-                    ),
-                    href="/",
-                    className="text-4xl p-4 text-center grow align-baseline",
-                ),
-                dcc.Link(
-                    "Config",
-                    href="/config",
-                    className="flex-none hover:bg-amber-100 p-4 align-middle",
-                ),
-            ],
-            className="bg-amber-300 flex flex-row shadow",
-        ),
-        dash.page_container,
-    ]
-)
-
-if __name__ == "__main__":
-    app.run(debug=True)

dashboard/components/lists.py

@@ -1,57 +0,0 @@
-from dash import dcc, html
-
-from ..libs.repository.repository import AbstractRepository
-
-
-def html_list_schema(stage:AbstractRepository, with_tables=True):
-    """ Build html list of schema in stage """
-    ul_classes = "ml-2"
-    schema_baseurl = f"/stg/{stage.name}/schema/"
-    if with_tables:
-        return html.Ul(
-            [
-                html.Li(
-                    children = [
-                        dcc.Link(
-                            schema,
-                            href=schema_baseurl + schema,
-                            className="text-lg hover:underline"
-                        ),
-                        html_list_table(stage, schema)
-                    ],
-                    className=""
-                ) for schema in stage.schemas()
-            ],
-            className=ul_classes
-        )
-    return html.Ul(
-        [
-            html.Li(
-                dcc.Link(
-                    schema,
-                    href=schema_baseurl + schema,
-                    className="text-lg hover:underline"
-                ),
-            ) for schema in stage.schemas()
-        ],
-        className=ul_classes
-    )
-
-
-def html_list_table(stage:AbstractRepository, schema:str):
-    """ Build html list of table in stage """
-    table_baseurl = f"/stg/{stage.name}/schm/{schema}/table/"
-    return html.Ul(
-        [
-            html.Li(
-                dcc.Link(
-                    table,
-                    href=table_baseurl + table,
-                    className="hover:underline"
-                ),
-            ) for table in stage.tables(schema=schema)
-        ],
-        className="ml-4"
-    )
-
-

dashboard/datalake.py

@@ -1,14 +0,0 @@
-from dotenv import dotenv_values
-
-from .libs.repository.fs_repository import FSRepository
-
-env = {
-    **dotenv_values(".env"),
-}
-
-stages = {
-    "raw": FSRepository("raw", f"{env['DATA_PATH']}/{env['RAW_SUBPATH']}"),
-    "staging": FSRepository("staging", f"{env['DATA_PATH']}/{env['STAGING_SUBPATH']}"),
-    "gold": FSRepository("gold", f"{env['DATA_PATH']}/{env['GOLD_SUBPATH']}"),
-    "mart": FSRepository("mart", f"{env['DATA_PATH']}/{env['MART_SUBPATH']}"),
-}

dashboard/libs/flux/flux.py

@@ -1,70 +0,0 @@
-from collections.abc import Callable
-from datetime import datetime
-
-import pandas as pd
-from pydantic import BaseModel
-
-from ..repository.repository import AbstractRepository
-
-
-class Schema(BaseModel):
-    repository: str
-    schema: str
-
-
-class Table(BaseModel):
-    repository: str
-    schema: str
-    table: str
-
-
-class Flux(BaseModel):
-    sources: list[Table]
-    destinations: dict[str, Table]
-    transformation: Callable[[list[pd.DataFrame]], dict[str, pd.DataFrame]]
-
-
-class State(BaseModel):
-    statuses: dict[str, dict]
-    qty_out: int
-    failed_lines: list[str]
-    start: datetime
-    end: datetime
-
-
-Repositories = dict[str, AbstractRepository]
-
-
-def open_source(repositories: Repositories, source: Table) -> pd.DataFrame:
-    return repositories[source.repository].read(source.table, source.schema)
-
-
-def write_source(
-    content: pd.DataFrame, repositories: Repositories, destination: Table
-) -> str:
-    return repositories[destination.repository].write(
-        content, destination.table, destination.schema
-    )
-
-
-def consume_flux(flux: Flux, repositories: dict[str, AbstractRepository]) -> State:
-    start = datetime.now()
-    src_dfs = [open_source(repositories, source) for source in flux.sources]
-
-    built_dfs = flux.transformation(src_dfs)
-
-    statuses = {
-        dest: write_source(df, repositories, flux.destinations[dest])
-        for dest, df in built_dfs.items()
-    }
-
-    end = datetime.now()
-    qty_out = 0
-    failed_lines = []
-    return State(
-        statuses=statuses,
-        qty_out=qty_out,
-        failed_lines=failed_lines,
-        start=start,
-        end=end,
-    )

dashboard/libs/repository/fs_repository.py

@@ -1,86 +0,0 @@
-from pathlib import Path
-
-import pandas as pd
-
-from .repository import AbstractRepository
-
-ACCEPTABLE_EXTENTIONS = {
-    "csv": [".csv"],
-    "excel": [".xls", ".xlsx"],
-}
-
-class FSRepository(AbstractRepository):
-    def __init__(self, name, basepath, metadata_engine=None):
-        self.name = name
-
-        self.basepath = Path(basepath)
-        assert self.basepath.exists()
-        self._metadata_engine = metadata_engine
-
-    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, recursive=True) -> list[str]:
-        return self.ls("", only_directories=True, recursive=True)
-
-    def tables(self, schema: str = ".") -> list[str]:
-        return self.ls(schema, only_files=True)
-
-    def build_table_path(self, table: str, schema: str):
-        table_path = self.basepath
-        if schema == ".":
-            return table_path / table
-        return table_path / schema / table
-
-    def infos(self, table: str, schema: str = "."):
-        table_path = self.build_table_path(table, schema)
-        pass
-
-    def read(self, table: str, schema: str = ".", **read_options):
-        table_path = self.build_table_path(table, schema)
-        assert table_path.exists()
-        extension = table_path.suffix
-        if extension in ACCEPTABLE_EXTENTIONS["csv"]:
-            return pd.read_csv(table_path, **read_options)
-
-        if extension in ACCEPTABLE_EXTENTIONS["excel"]:
-            return pd.read_excel(table_path, engine = "openpyxl", **read_options)
-
-        raise ValueError("Bad extention. Can't open the table.")
-
-    def write(self, content, table: str, schema: str = "."):
-        table_path = self.build_table_path(table, schema)
-        pass
-
-    def delete_table(self, table: str, schema: str = "."):
-        table_path = self.build_table_path(table, schema)
-        pass

dashboard/libs/repository/metadata.py

@@ -1,5 +0,0 @@
-from abc import ABC
-
-
-class AbstractMetadataEngine(ABC):
-    pass

dashboard/libs/repository/repository.py

@@ -1,37 +0,0 @@
-import abc
-
-from .metadata import AbstractMetadataEngine
-
-
-class AbstractRepository(abc.ABC):
-    metadata_engine = AbstractMetadataEngine
-
-    @abc.abstractmethod
-    def schemas(self) -> list[str]:
-        """List schemas"""
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def tables(self, schema) -> list[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
-
-    @abc.abstractmethod
-    def read(self, table: str, schema: str):
-        """Get content of the table"""
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def write(self, content, table: str, schema: str):
-        """Write content into the table"""
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def delete_table(self, table: str, schema: str):
-        """Delete the table"""
-        raise NotImplementedError

dashboard/pages/config.py

@@ -1,14 +0,0 @@
-from dash import html
-from dotenv import dotenv_values
-import os
-
-env = {
-    **dotenv_values(".env"),
-    **os.environ,
-}
-
-
-layout = html.Div([
-    html.H1('This is our Config page'),
-    html.Ul(children = [html.Li(f"{k} = {v}") for k,v in env.items()]),
-])

dashboard/pages/home.py

@@ -1,27 +0,0 @@
-from dash import dcc, html
-
-from ..components.lists import html_list_schema
-from ..datalake import stages
-
-layout = html.Div([
-    html.Div(children=[
-        html.Ul(
-            children=[
-                html.Li(
-                    children=[
-                        dcc.Link(
-                            stagename,
-                            href=f"/stage/{stagename}",
-                            className="text-2xl text-center p-2 bg-amber-100 rounded shadow"
-                        ),
-                        html_list_schema(stage)
-                    ],
-                    className="flex-1 bg-gray-100 rounded flex flex-col shadow"
-                ) for stagename, stage in stages.items()
-            ],
-            className="flex flex-row space-x-2"
-        )
-    ],
-    className="w-full mt-4 px-2"
-    ),
-])

dashboard/pages/repository.py

@@ -1,18 +0,0 @@
-from dash import html
-
-from ..components.lists import html_list_schema
-from ..libs.repository.repository import AbstractRepository
-
-
-def layout_factory(repositories: dict[str, AbstractRepository]):
-    def layout(repository_name: str = ""):
-        repository = repositories[repository_name]
-        return html.Div(
-            [
-                html.H2(f"{repository.name}", className="text-2xl p-4 py-2"),
-                html_list_schema(repository),
-            ],
-            className="flex flex-col",
-        )
-
-    return layout

dashboard/pages/schema.py

@@ -1,28 +0,0 @@
-from dash import dcc, html
-
-from ..libs.repository.repository import AbstractRepository
-
-
-def layout_factory(repositories: dict[str, AbstractRepository]):
-    def layout(repository_name: str = "", schema_name: str = ""):
-        repository = repositories[repository_name]
-        return html.Div(
-            [
-                html.H2(
-                    [
-                        dcc.Link(
-                            f"{repository.name}",
-                            href=f"/repository/{repository.name}",
-                            className="hover:underline",
-                        ),
-                        html.Span(" > "),
-                        html.Span(
-                            f"{schema_name}",
-                        ),
-                    ],
-                    className="text-2xl p-4 py-2",
-                ),
-            ]
-        )
-
-    return layout

dashboard/pages/table.py

@@ -1,130 +0,0 @@
-from dash import Input, Output, State, dash_table, dcc, html
-from dash.exceptions import PreventUpdate
-
-from ..libs.repository.repository import AbstractRepository
-
-
-def layout_factory(repositories: dict[str,AbstractRepository]):
-    def layout(repository_name:str="", schema_name:str="", table_name:str=""):
-        repository = repositories[repository_name]
-        df = repository.read(table=table_name, schema=schema_name)
-        return html.Div([
-            dcc.Store(id="table_backup"),
-            html.Div([
-                html.H2([
-                    dcc.Link(
-                        f"{repository.name}",
-                        href=f"/repository/{repository.name}",
-                        className="hover:underline"
-                    ),
-                    html.Span(" > "),
-                    dcc.Link(
-                        f"{schema_name}",
-                        href=f"/stg/{repository.name}/schema/{schema_name}",
-                        className="hover:underline"
-                    ),
-                    html.Span(" > "),
-                    html.Span(table_name),
-                ],
-                className="text-2xl"
-                ),
-                html.Div([
-                    html.Button(
-                        "Editer",
-                        id="btn_edit",
-                        className="rounded border px-2 py-1",
-                        style={"display": "block"}
-                    ),
-                    html.Button(
-                        "Sauver",
-                        id="btn_save",
-                        className="rounded border px-2 py-1 border-green-500 hover:bg-green-500",
-                        style={"display": "none"}
-                    ),
-                    html.Button(
-                        "Annuler",
-                        id="btn_cancel",
-                        className="rounded border px-2 py-1 border-red-500 hover:bg-red-500",
-                        style={"display": "none"}
-                    ),
-                ],
-                className="flex flex-row space-x-2",
-                id="toolbar"
-                ),
-            ],
-            className="flex flex-row justify-between p-4"
-            ),
-            html.Div([
-                html.Div([
-                    dash_table.DataTable(
-                        id="datatable",
-                        data=df.to_dict('records'),
-                        columns=[{"name": i, "id": i} for i in df.columns],
-                        filter_action="native",
-                        sort_action="native",
-                        sort_mode="multi",
-                        editable=False
-                    )
-                ])
-            ],
-         className="overflow-y-auto"
-         ),
-        ],
-        className="p-2"
-        )
-    return layout
-
-
-def callback_factory(app):
-    @app.callback(
-        Output("datatable", 'editable', allow_duplicate=True),
-        Output("table_backup", 'data'),
-        Input("btn_edit", "n_clicks"),
-        State("datatable", 'data'),
-        prevent_initial_call=True
-    )
-    def activate_editable(n_clicks, df_src):
-        if n_clicks is None:
-            raise PreventUpdate
-        if n_clicks > 0:
-            df_backup = df_src.copy()
-            return True, df_backup
-        raise PreventUpdate
-
-    @app.callback(
-        Output("datatable", 'editable', allow_duplicate=True),
-        Output("datatable", 'data', allow_duplicate=True),
-        Input("btn_cancel", "n_clicks"),
-        State("table_backup", 'data'),
-        prevent_initial_call=True
-    )
-    def cancel_modifications(n_clicks, data):
-        if n_clicks is None:
-            raise PreventUpdate
-        if n_clicks > 0 and data is not None:
-            return False, data.copy()
-        raise PreventUpdate
-
-    @app.callback(
-        Output("datatable", 'editable'),
-        Output("datatable", 'data'),
-        Input("btn_save", "n_clicks"),
-        State("datatable", 'editable'),
-    )
-    def save_modifications(n_clicks, editable):
-        if n_clicks is None:
-            raise PreventUpdate
-        if n_clicks > 0:
-            return not editable
-        return editable
-
-    @app.callback(
-        Output("btn_edit", "style"),
-        Output("btn_save", "style"),
-        Output("btn_cancel", "style"),
-        Input("datatable", "editable"),
-    )
-    def toolbar(editable):
-        if editable:
-            return {"display": "none"}, {"display": "block"}, {"display": "block"}
-        return {"display": "block"}, {"display": "none"}, {"display": "none"}

plesna/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_set.py

@@ -0,0 +1,33 @@
+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

requirements.txt

@@ -1,6 +1,7 @@
 jupyter==1.0.0 
-pandas==2.2.2
+pandas==1.5.0
-pydantic==2.8.2
+pdf-oralia==0.3.11
+pydantic==2.6.1
 click==8.1.7
-openpyxl==3.1.5
+dlt[duckdb]>=0.4.3a0
-xlrd==2.0.1
+openpyxl>=3.0.0

tests/graphs/test_graph.py

@@ -0,0 +1,107 @@
+import pytest
+
+from plesna.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_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/fs_examples/username.csv

@@ -1,7 +0,0 @@
-Username;Identifier;First name;Last name
-booker12;9012;Rachel;Booker
-grey07;2070;Laura;Grey
-johnson81;4081;Craig;Johnson
-jenkins46;9346;Mary;Jenkins
-smith79;5079;Jamie;Smith
-

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

tests/test_flux.py

@@ -1,131 +0,0 @@
-import pandas as pd
-import pytest
-
-from dashboard.libs.flux.flux import Flux, consume_flux
-from dashboard.libs.repository.repository import AbstractRepository
-
-FakeTable = pd.DataFrame
-FakeSchema = dict[str, pd.DataFrame]
-FakeSchemas = dict[str, FakeSchema]
-
-
-class FakeRepository(AbstractRepository):
-    def __init__(self, schemas: FakeSchemas):
-        self._schemas = {}
-        for schema_name, tables in schemas.items():
-            schema = {}
-            for table, df in tables.items():
-                schema[table] = {
-                    "df": df,
-                    "metadata": {
-                        "status": "new",
-                        "qty_read": 0,
-                        "qty_write": 0,
-                    },
-                }
-            self._schemas[schema_name] = schema
-
-    def schemas(self):
-        """List schemas"""
-        return list(self._schemas.keys())
-
-    def tables(self, schema):
-        """List table's name in schema"""
-        return list(self._schemas[schema].keys())
-
-    def infos(self, table: str, schema: str) -> dict[str, str]:
-        """Get infos about the table"""
-        return self._schemas[schema][table]["metadata"]
-
-    def read(self, table, schema) -> pd.DataFrame:
-        """Get content of the table"""
-        self._schemas[schema][table]["metadata"]["qty_read"] += 1
-        return self._schemas[schema][table]["df"]
-
-    def write(self, content, table, schema) -> dict[str, str]:
-        """Write content into the table"""
-        try:
-            self._schemas[schema][table]["df"] = content
-        except KeyError:
-            self._schemas[schema][table] = {
-                "df": content,
-                "metadata": {
-                    "status": "new",
-                    "qty_read": 0,
-                    "qty_write": 0,
-                },
-            }
-        self._schemas[schema][table]["metadata"]["status"] = "modified"
-        self._schemas[schema][table]["metadata"]["qty_write"] += 1
-        return self.infos(table, schema)
-
-    def delete_table(self, table, schema):
-        """Delete the table"""
-        raise NotImplementedError
-
-
-def test_fakerepository():
-    fakerepository = FakeRepository(
-        {
-            "foo": {
-                "table1": pd.DataFrame({"A": []}),
-                "table2": pd.DataFrame({"B": []}),
-            },
-            "bar": {
-                "table1": pd.DataFrame({"C": []}),
-                "table2": pd.DataFrame({"D": []}),
-            },
-        }
-    )
-    assert fakerepository.schemas() == ["foo", "bar"]
-    assert fakerepository.tables("foo") == ["table1", "table2"]
-    assert fakerepository.infos("table1", "foo") == {
-        "status": "new",
-        "qty_read": 0,
-        "qty_write": 0,
-    }
-    assert fakerepository.read("table1", "foo").equals(pd.DataFrame({"A": []}))
-    assert fakerepository.infos("table1", "foo") == {
-        "status": "new",
-        "qty_read": 1,
-        "qty_write": 0,
-    }
-
-    df = pd.DataFrame({"A": [1, 2]})
-    assert fakerepository.write(df, "table1", "foo") == {
-        "status": "modified",
-        "qty_read": 1,
-        "qty_write": 1,
-    }
-
-
-def test_consume_flux():
-    source_repository = FakeRepository(
-        {
-            "source": {
-                "table1": pd.DataFrame({"A": [1, 2, 3]}),
-            },
-        }
-    )
-    dest_repository = FakeRepository(
-        {
-            "destination": {},
-        }
-    )
-    repositories = {
-        "source": source_repository,
-        "dest": dest_repository,
-    }
-    transformation = lambda dfs: {"dest": dfs[0] * 2}
-
-    flux = Flux(
-        sources=[{"repository": "source", "schema": "source", "table": "table1"}],
-        destinations={
-            "dest": {"repository": "dest", "schema": "destination", "table": "table1"}
-        },
-        transformation=transformation,
-    )
-
-    state = consume_flux(flux, repositories)
-    assert state.statuses["dest"] == {'status': 'modified', 'qty_read': 0, 'qty_write': 1}
-    assert dest_repository.read("table1", "destination").equals(pd.DataFrame({"A": [2, 4, 6]}))