Feat: add repository to dataplatform

Makefile

@@ -1,66 +0,0 @@
-DATA_BASE=./datas
-
-PDF_BASE=$(DATA_BASE)/pdfs
-PDF_YEARS=$(wildcard $(PDF_BASE)/*)
-
-RAW_BASE=$(DATA_BASE)/raw
-RAW_CRG=$(RAW_BASE)/CRG
-RAW_CRG_YEARS=$(subst $(PDF_BASE), $(RAW_CRG), $(PDF_YEARS))
-
-
-$(RAW_CRG)/%/: $(wildcard $(PDF_BASE)/%/*)
-	echo $(wildcard $(PDF_BASE)/$*/*)
-	@echo ----
-	ls $(PDF_BASE)/$*/
-	@echo ----
-	echo $*
-	@echo ----
-	echo $^
-	@echo ----
-	echo $?
-
-#./datas/raw/CRG/%: 
-#pdf-oralia extract all --src $$year --dest $$(subst $$PDF_BASE, $$RAW_CRG, $$year)
-# $(RAW_CRG_YEARS): $(PDF_PATHS)
-# 	for year in $(PDF_PATHS); do \
-# 		echo $$year; \
-# 		echo $$(subst $$PDF_BASE, $$RAW_CRG, $$year); \
-# 		echo "----"; \
-# 	done;
-
-extract_pdfs:
-	for year in 2021 2022 2023 2024; do \
-		mkdir -p $(RAW_CRG)/$$year/extracted;\
-		pdf-oralia extract all --src $(PDF_BASE)/$$year/ --dest $(RAW_CRG)/$$year/extracted; \
-		pdf-oralia join --src $(RAW_CRG)/$$year/extracted/ --dest $(RAW_CRG)/$$year/; \
-	done
-
-clean_raw:
-	rm -rf ./PLESNA Compta SYSTEM/raw/**/*.csv
-
-clean_built:
-	rm -rf $(DATA_BASE)/staging/**/*.csv
-	rm -rf $(DATA_BASE)/gold/**/*.csv
-	rm -rf $(DATA_BASE)/datamart/**/*.csv
-	rm -rf $(DATA_BASE)/datamart/**/*.xlsx
-
-run_ingest:
-	python -m scripts ingest
-
-run_feature:
-	python -m scripts feature
-
-run_datamart:
-	python -m scripts datamart
-
-build: clean_built run_ingest run_feature run_datamart
-
-clean_all: clean_built clean_raw
-
-import_nextcloud:
-	rsync -av ~/Nextcloud/PLESNA\ Compta\ SYSTEM/Histoire/ ./datas/Histoire
-
-push_nextcloud:
-	rsync -av ./datas/datamart/ ~/Nextcloud/PLESNA\ Compta\ SYSTEM/DataMart
-
-

README.md

@@ -1,5 +1,15 @@
 # E(T)LT pour Plesna
 
+## Installation
+
+## Concepts
+
+- `dataplatform`: agrégation d'un datacatalogue, de moteur de compute et du dag des transformations.
+- `datacatalogue`: gestion du contenu des datastores.
+- `datastore`: interface de stockage des données.
+- `compute`: moteur de traitement des fluxs.
+- `graph/dag`: organisation logique des fluxs et des données.
+
 ## Stages
 
 - Raw: fichiers les plus brutes possibles

dashboard/app.py

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

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

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

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

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

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

dashboard/libs/repository/repository.py

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

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

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

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

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

@@ -0,0 +1,130 @@
+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/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/dataplatform.py

@@ -0,0 +1,27 @@
+from plesna.graph.graph_set import GraphSet
+from plesna.storage.repository.repository import Repository
+
+
+class DataPlateformError(Exception):
+    pass
+
+
+class DataPlateform:
+    def __init__(self):
+        self._graphset = GraphSet()
+        self._metadata_engine = ""
+        self._transformations = {}
+        self._repositories = {}
+
+    def add_repository(self, name: str, repository: Repository):
+        if name in self._repositories:
+            raise DataPlateformError("The repository {name} already exists")
+
+        self._repositories[name] = repository
+
+    @property
+    def repositories(self) -> list[str]:
+        return list(self._repositories)
+
+    def repository(self, name: str) -> Repository:
+        return self._repositories[name]

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,55 @@
+from pydantic import BaseModel
+
+
+class Schema(BaseModel):
+    """Where multiple tables are stored
+
+    id: uniq identifier for the schema
+    repo_id: id of the repo where the schema belong to
+    name: name of the schema
+    value: string which describe where to find the schema in the repository
+    """
+
+    id: str
+    repo_id: str
+    name: str
+    value: str
+    tables: list[str] = []
+
+
+class Table(BaseModel):
+    """Place where same structured data are stored
+
+    id: uniq identifier for the table
+    repo_id: id of the repo where the table belong to
+    schema_id: id of the schema where table belong to
+    name: the name of the table
+    value: string which describe where to find the table in the storage system
+    """
+
+    id: str
+    repo_id: str
+    schema_id: str
+    name: str
+    value: str
+    partitions: list[str] = []
+
+
+class Partition(BaseModel):
+    """Place where data are stored
+
+    id: uniq identifier for the table
+    repo_id: id of the repo where the table belong to
+    schema_id: id of the schema where table belong to
+    table_id: id of the schema where table belong to
+    name: the name of the partition
+    value: string which describe where to find the partition in the storage system
+
+    """
+
+    id: str
+    repo_id: str
+    schema_id: str
+    table_id: str
+    name: 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 = {}

plesna/storage/datacatalogue.py

@@ -0,0 +1,24 @@
+import abc
+
+from plesna.models.storage import Schema
+
+
+class DataCatalogue:
+    def __init__(self):
+        pass
+
+    @property
+    @abc.abstractmethod
+    def schemas(self) -> list[str]:
+        """List schema's names"""
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def schema(self, name: str) -> Schema:
+        """Get the schema properties"""
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def tables(self, schema: str) -> list[str]:
+        """List table's name in schema"""
+        raise NotImplementedError

plesna/storage/fake_datacatalogue.py

@@ -0,0 +1,81 @@
+from pathlib import Path
+
+from pydantic import BaseModel, computed_field
+
+from plesna.models.storage import Schema, Table
+
+from .datacatalogue import DataCatalogue
+
+
+class FakeSchema(BaseModel):
+    name: str
+
+    @computed_field
+    @property
+    def ref(self) -> Schema:
+        return Schema(
+            id=str(self.name),
+            value=str(self.name),
+        )
+
+
+class FakeTable(BaseModel):
+    name: str
+    data: dict[str, list]
+
+    @computed_field
+    @property
+    def ref(self) -> Table:
+        return Table(
+            id=str(self.name),
+            value=str(self.name),
+        )
+
+
+class FakeDataCatalogue(DataCatalogue):
+    """DataCatalogue based on dictionnaries"""
+
+    def __init__(self, name: str):
+        self.name = name
+
+    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/storage/repository/fs_repository.py

@@ -0,0 +1,152 @@
+from pathlib import Path
+
+from pydantic import BaseModel, computed_field
+
+from plesna.models.storage import Partition, Schema, Table
+from plesna.storage.repository.repository import Repository
+
+
+class FSPartition(BaseModel):
+    name: str
+    path: Path
+
+    @computed_field
+    @property
+    def ref(self) -> Partition:
+        return Partition(
+            id=str(self.path),
+            repo_id=str(self.path.parent.parent.parent),
+            schema_id=str(self.path.parent.parent),
+            table_id=str(self.path.parent),
+            name=self.name,
+            value=str(self.path.absolute()),
+        )
+
+
+class FSTable(BaseModel):
+    name: str
+    path: Path
+    is_partitionned: bool
+    partitions: list[str] = []
+
+    @computed_field
+    @property
+    def ref(self) -> Table:
+        return Table(
+            id=str(self.path),
+            repo_id=str(self.path.parent.parent),
+            schema_id=str(self.path.parent),
+            name=self.name,
+            value=str(self.path.absolute()),
+            partitions=self.partitions,
+        )
+
+
+class FSSchema(BaseModel):
+    name: str
+    path: Path
+    tables: list[str]
+
+    @computed_field
+    @property
+    def ref(self) -> Schema:
+        return Schema(
+            id=str(self.path),
+            repo_id=str(self.path.parent),
+            name=self.name,
+            value=str(self.path.absolute()),
+            tables=self.tables,
+        )
+
+
+class FSRepository(Repository):
+    """Repository based on files tree structure
+
+    - first level: schemas
+    - second level: tables
+    - third level: partition (actual datas)
+
+    """
+
+    def __init__(self, name: str, basepath: str, id: str):
+        self._basepath = Path(basepath)
+        self.name = name
+        self.id = id
+
+        assert self._basepath.exists()
+
+    def ls(
+        self, dir="", only_files=False, only_directories=False, recursive=False
+    ) -> list[str]:
+        """List files in dir
+
+        :param dir: relative path from self._basepath
+        :param only_files: if true return only files
+        :param only_directories: if true return only directories
+        :param recursive: list content recursively (only for)
+        :return: list of string describing path from self._basepath / dir
+        """
+        dirpath = self._basepath / dir
+
+        if recursive:
+            paths = dirpath.rglob("*")
+        else:
+            paths = dirpath.iterdir()
+
+        if only_files:
+            return [
+                str(f.relative_to(dirpath))
+                for f in paths
+                if not f.is_dir() and not str(f).startswith(".")
+            ]
+        if only_directories:
+            return [
+                str(f.relative_to(dirpath))
+                for f in paths
+                if f.is_dir() and not str(f).startswith(".")
+            ]
+
+        return [
+            str(f.relative_to(dirpath)) for f in paths if not str(f).startswith(".")
+        ]
+
+    def schemas(self) -> list[str]:
+        """List schemas (sub directories within basepath)"""
+        subdirectories = self.ls("", only_directories=True)
+        return [str(d) for d in subdirectories]
+
+    def _schema(self, name: str) -> FSSchema:
+        """List schemas (sub directories within basepath)"""
+        schema_path = self._basepath / name
+        tables = self.ls(name)
+        return FSSchema(name=name, path=schema_path, tables=tables)
+
+    def schema(self, name: str) -> Schema:
+        return self._schema(name).ref
+
+    def _table(self, schema: str, name: str) -> FSTable:
+        """Get infos on the table"""
+        table_path = self._basepath / schema / name
+        is_partitionned = table_path.is_dir()
+        if is_partitionned:
+            partitions = self.ls(f"{schema}/{name}", only_files=True)
+        else:
+            partitions = []
+
+        return FSTable(
+            name=name,
+            path=table_path,
+            is_partitionned=is_partitionned,
+            partitions=partitions,
+        )
+
+    def table(self, schema: str, name: str) -> Table:
+        return self._table(schema, name).ref
+
+    def _partition(self, schema: str, table: str, partition: str) -> FSPartition:
+        """Get infos on the partition"""
+        table_path = self._basepath / schema / table
+        return FSPartition(name=partition, table_path=table_path)
+
+    def partition(self, schema: str, name: str) -> Partition:
+        return self._partition(schema, name).ref

plesna/storage/repository/repository.py

@@ -0,0 +1,38 @@
+import abc
+
+from plesna.models.storage import Partition, Schema, Table
+
+
+class Repository:
+    def __init__(self):
+        pass
+
+    @abc.abstractmethod
+    def schemas(self) -> list[str]:
+        """List schema's names"""
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def schema(self, name: str) -> Schema:
+        """Get the schema properties"""
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def tables(self, schema: str) -> list[str]:
+        """List table's name in schema"""
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def table(self, schema: str, name: str) -> Table:
+        """Get the table properties"""
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def partitions(self, schema: str, table: str) -> list[str]:
+        """List partition's name in table"""
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def partition(self, schema: str, name: str, partition: str) -> Partition:
+        """Get the partition properties"""
+        raise NotImplementedError

requirements.txt

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

tests/compute/test_consume_flux.py

@@ -0,0 +1,43 @@
+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", repo_id="test", schema_id="test", name="test", value="here"
+        ),
+        "src2": Table(
+            id="src2", repo_id="test", schema_id="test", name="test", value="here"
+        ),
+    }
+    targets = {
+        "tgt1": Table(
+            id="tgt1", repo_id="test", schema_id="test", name="test", value="this"
+        ),
+        "tgt2": Table(
+            id="tgt2", repo_id="test", schema_id="test", name="test", 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/dataplatform/test_dataplateform.py

@@ -0,0 +1,74 @@
+import shutil
+from pathlib import Path
+
+import pytest
+
+from plesna.dataplatform import DataPlateform
+from plesna.storage.repository.fs_repository import FSRepository
+
+FIXTURE_DIR = Path(__file__).parent.parent / Path("raw_datas")
+
+
+@pytest.fixture
+def repository(tmp_path) -> FSRepository:
+    raw_path = Path(tmp_path) / "raw"
+    raw_path.mkdir()
+
+    example_src = FIXTURE_DIR
+    assert example_src.exists()
+
+    recovery_loc = raw_path / "recovery"
+    recovery_loc.mkdir()
+    username_loc = raw_path / "username"
+    username_loc.mkdir()
+    salary_loc = raw_path / "salary"
+    salary_loc.mkdir()
+
+    for f in example_src.glob("*"):
+        if "recovery" in str(f):
+            shutil.copy(f, recovery_loc)
+        if "salary" in str(f):
+            shutil.copy(f, salary_loc)
+        else:
+            shutil.copy(f, username_loc)
+
+    bronze_path = Path(tmp_path) / "bronze"
+    bronze_path.mkdir()
+    silver_path = Path(tmp_path) / "silver"
+    silver_path.mkdir()
+
+    return FSRepository("test", tmp_path, "test")
+
+
+def test_add_repository(
+    repository: FSRepository,
+):
+    dp = DataPlateform()
+    dp.add_repository("test", repository)
+
+    assert dp.repositories == ["test"]
+
+    assert dp.repository("test") == repository
+
+
+@pytest.fixture
+def dataplatform(
+    repository: FSRepository,
+) -> DataPlateform:
+    dp = DataPlateform()
+    dp.add_repository("test", repository)
+    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_add_flux(dataplatform: DataPlateform):
+    # dataplatform.add_flux()
+    pass

tests/e2e/raw_data/username.csv

@@ -0,0 +1,7 @@
+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/e2e/test_datalake.py

@@ -0,0 +1,39 @@
+from pathlib import Path
+
+import pytest
+
+from plesna.dataplatform import DataPlateform
+from plesna.datastore.fs_datacatalogue import FSDataCatalogue
+
+FIXTURE_DIR = Path(__file__).parent / Path("raw_data")
+
+
+@pytest.fixture
+def raw_catalogue(tmp_path):
+    raw_path = Path(tmp_path) / "raw"
+    return FSDataCatalogue(raw_path)
+
+
+@pytest.fixture
+def bronze_catalogue(tmp_path):
+    bronze_path = Path(tmp_path) / "bronze"
+    return FSDataCatalogue(bronze_path)
+
+
+@pytest.fixture
+def silver_catalogue(tmp_path):
+    silver_path = Path(tmp_path) / "silver"
+    return FSDataCatalogue(silver_path)
+
+
+@pytest.fixture
+def dataplateform(
+    raw_catalogue: FSDataCatalogue,
+    bronze_catalogue: FSDataCatalogue,
+    silver_catalogue: FSDataCatalogue,
+):
+    dp = DataPlateform()
+    dp.add_datacatalague("raw", raw_catalogue)
+    dp.add_datacatalague("bronze", bronze_catalogue)
+    dp.add_datacatalague("silver", silver_catalogue)
+    pass

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/raw_datas/username.csv

@@ -0,0 +1,7 @@
+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/storage/test_fs_repository.py

@@ -0,0 +1,77 @@
+import shutil
+from pathlib import Path
+
+import pytest
+
+from plesna.models.storage import Schema
+from plesna.storage.repository.fs_repository import FSRepository
+
+FIXTURE_DIR = Path(__file__).parent.parent / Path("./raw_datas/")
+
+
+@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 = FSRepository("example", location, "example")
+    assert repo.ls() == [
+        "username",
+        "salary",
+    ]
+
+    assert repo.ls(recursive=True) == [
+        "username",
+        "salary",
+        "username/username.csv",
+        "username/username-password-recovery-code.xlsx",
+        "username/username-password-recovery-code.xls",
+        "salary/salary.pdf",
+    ]
+
+
+@pytest.fixture
+def repository(location) -> FSRepository:
+    return FSRepository("example", location, "example")
+
+
+def test_list_schema(location, repository):
+    assert repository.schemas() == ["username", "salary"]
+    assert repository.schema("username").name == "username"
+    assert repository.schema("username").id == str(location / "username")
+    assert repository.schema("username").repo_id == str(location)
+    assert repository.schema("username").value == str(location / "username")
+
+
+def test_list_tables_schema(repository):
+    assert repository.schema("username").tables == [
+        "username.csv",
+        "username-password-recovery-code.xlsx",
+        "username-password-recovery-code.xls",
+    ]
+    assert repository.schema("salary").tables == ["salary.pdf"]
+
+
+def test_describe_table(location, repository):
+    table = repository.table("username", "username.csv")
+    assert table.id == str(location / "username" / "username.csv")
+    assert table.repo_id == str(location)
+    assert table.schema_id == str(location / "username")
+    assert table.name == "username.csv"
+    assert table.value == str(location / "username" / "username.csv")
+    assert table.partitions == []

tests/test_flux.py

@@ -0,0 +1,131 @@
+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]}))

uv.lock

@@ -0,0 +1,7 @@
+version = 1
+requires-python = ">=3.13"
+
+[[package]]
+name = "plesna"
+version = "0.1.0"
+source = { virtual = "." }