Feat: test on pandas xlsx and ods file reader

dashboard/app.py

@@ -1,60 +1,64 @@
 import dash
-from dash import Dash, html, dcc
+from dash import Dash, dcc, html
-from .pages import home, config, stage, schema, table
 
-external_scripts = [
+from .datalake import stages
-    {'src': 'https://cdn.tailwindcss.com'}
+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__,
+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='/', 
+    path="/",
     layout=home.layout,
 )
+dash.register_page(config.__name__, path="/config", layout=config.layout)
 dash.register_page(
-    config.__name__, 
+    repository.__name__,
-    path='/config', 
+    path_template="/repository/<repository_name>",
-    layout=config.layout
+    layout=repository.layout_factory(stages),
-)
-dash.register_page(
-    stage.__name__, 
-    path_template='/stage/<stage_name>', 
-    layout=stage.layout
 )
 dash.register_page(
     schema.__name__,
-    path_template='/stg/<stage_name>/schema/<schema_name>',
+    path_template="/stg/<repository_name>/schema/<schema_name>",
-    layout=schema.layout
+    layout=schema.layout_factory(stages),
 )
 dash.register_page(
     table.__name__,
-    path_template='/stg/<stage_name>/schm/<schema_name>/table/<table_name>',
+    path_template="/stg/<repository_name>/schm/<schema_name>/table/<table_name>",
-    layout=table.layout
+    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,
+    ]
 )
 
-app.layout = html.Div([
+if __name__ == "__main__":
-    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,7 +1,9 @@
-from ..libs.stage.stage import AbstractStage
+from dash import dcc, html
-from dash import html, dcc
 
-def html_list_schema(stage:AbstractStage, with_tables=True):
+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/"
@@ -36,7 +38,7 @@ def html_list_schema(stage:AbstractStage, with_tables=True):
     )
 
 
-def html_list_table(stage:AbstractStage, schema:str):
+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(

dashboard/datalake.py

@@ -1,13 +1,14 @@
-from .libs.stage.fs_stage import FSStage
 from dotenv import dotenv_values
 
+from .libs.repository.fs_repository import FSRepository
+
 env = {
     **dotenv_values(".env"),
 }
 
 stages = {
-    "raw": FSStage("raw", f"{env['DATA_PATH']}/{env['RAW_SUBPATH']}"),
+    "raw": FSRepository("raw", f"{env['DATA_PATH']}/{env['RAW_SUBPATH']}"),
-    "staging": FSStage("staging", f"{env['DATA_PATH']}/{env['STAGING_SUBPATH']}"),
+    "staging": FSRepository("staging", f"{env['DATA_PATH']}/{env['STAGING_SUBPATH']}"),
-    "gold": FSStage("gold", f"{env['DATA_PATH']}/{env['GOLD_SUBPATH']}"),
+    "gold": FSRepository("gold", f"{env['DATA_PATH']}/{env['GOLD_SUBPATH']}"),
-    "mart": FSStage("mart", f"{env['DATA_PATH']}/{env['MART_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/fs_schema.py

@@ -1,35 +0,0 @@
-from .schema import AbstractSchema
-from pathlib import Path
-
-class FSSchema(AbstractSchema):
-    def __init__(self, basepath, metadata_engine=None):
-        self.basepath = basepath
-        self._metadata_engine = metadata_engine
-
-    def ls(self, dir, only_files=True):
-        dirpath = Path(dir)
-        if only_files:
-            return [f for f in dirpath.iterdir() if f.is_dir()]
-        return [f for f in dirpath.iterdir()]
-
-    def tables(self, dir, only_files=True):
-        dirpath = Path(dir)
-        if only_files:
-            return [f for f in dirpath.iterdir() if f.is_dir()]
-        return [f for f in dirpath.iterdir()]
-
-    def info(self, path):
-        path = Path(path)
-        pass
-
-    def read(self, path):
-        path = Path(path)
-        pass
-
-    def write(self, path, content):
-        path = Path(path)
-        pass
-
-    def delete(self, path):
-        path = Path(path)
-        pass

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/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/libs/stage/fs_stage.py

@@ -1,61 +0,0 @@
-from .stage import AbstractStage
-from pathlib import Path
-import pandas as pd
-
-class FSStage(AbstractStage):
-    def __init__(self, name, basepath, metadata_engine=None):
-        self.name = name
-
-        self.basepath = Path(basepath)
-        self._metadata_engine = metadata_engine
-
-    def ls(self, dir, only_files=False, only_directories=False, recursive=False) -> list[str]:
-        dirpath = Path(dir)
-
-        if only_files:
-            return [str(f.relative_to(dirpath)) for f in dirpath.iterdir() if not f.is_dir()]
-
-        if only_directories:
-            if recursive:
-                return [str(f[0].relative_to(dirpath)) for f in dirpath.walk()]
-
-            return [str(f.relative_to(dirpath)) for f in dirpath.iterdir() if f.is_dir()]
-
-        return [str(f.relative_to(dirpath)) for f in dirpath.iterdir()]
-
-    def schemas(self, recursive=True) -> list[str]:
-        dirpath = self.basepath
-        return self.ls(dirpath, only_directories=True, recursive=True)
-
-    def tables(self, schema:str) -> list[str]:
-        dirpath = self.basepath / schema
-        return self.ls(dirpath, 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 info(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)
-        extension = table_path.suffix
-        if extension == '.csv':
-            return pd.read_csv(table_path, **read_options)
-
-        if extension == '.xlsx':
-            return pd.read_excel(table_path, **read_options)
-
-        raise ValueError("Can't open the table")
-
-    def write(self, table:str, content, schema:str='.'):
-        table_path = self.build_table_path(table, schema)
-        pass
-
-    def delete(self, table:str, schema:str='.'):
-        table_path = self.build_table_path(table, schema)
-        pass

dashboard/libs/stage/stage.py

@@ -1,36 +0,0 @@
-import abc
-from .metadata import AbstractMetadataEngine
-
-
-class AbstractStage(abc.ABC):
-    metadata_engine = AbstractMetadataEngine
-
-    @abc.abstractmethod
-    def schemas():
-        """ List schemas """
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def tables(schema):
-        """ List table in schema"""
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def info(self, path):
-        """ Get infos about a file"""
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def read(self, path):
-        """ Get content of a file"""
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def write(self, path, content):
-        """ Write content into the file"""
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def delete(self, path):
-        """ Delete the file """
-        raise NotImplementedError

dashboard/pages/home.py

@@ -1,7 +1,7 @@
-from dash import html, dcc
+from dash import dcc, html
-from ..datalake import stages
-from ..components.lists import html_list_schema
 
+from ..components.lists import html_list_schema
+from ..datalake import stages
 
 layout = html.Div([
     html.Div(children=[

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

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

dashboard/pages/stage.py

@@ -1,18 +0,0 @@
-from dash import html, dcc
-from ..datalake import stages
-from ..libs.stage.stage import AbstractStage
-from ..components.lists import html_list_schema
-
-
-def layout(stage_name=None):
-    stage = stages[stage_name]
-    return html.Div([
-
-        html.H2(
-            f"{stage.name}",
-            className="text-2xl p-4 py-2"
-        ),
-            html_list_schema(stage)
-        ],
-        className = "flex flex-col"
-    )

dashboard/pages/table.py

@@ -1,25 +1,26 @@
-from dash import html, dcc, dash_table, callback, Input, Output, State
+from dash import Input, Output, State, dash_table, dcc, html
 from dash.exceptions import PreventUpdate
-from ..datalake import stages
+
-from ..libs.stage.stage import AbstractStage
+from ..libs.repository.repository import AbstractRepository
 
 
-def layout(stage_name=None, schema_name=None, table_name=None):
+def layout_factory(repositories: dict[str,AbstractRepository]):
-    stage = stages[stage_name]
+    def layout(repository_name:str="", schema_name:str="", table_name:str=""):
-    df = stage.read(table=table_name, schema=schema_name)
+        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"{stage.name}",
+                        f"{repository.name}",
-                    href=f"/stage/{stage.name}",
+                        href=f"/repository/{repository.name}",
                         className="hover:underline"
                     ),
                     html.Span(" > "),
                     dcc.Link(
                         f"{schema_name}",
-                    href=f"/stg/{stage.name}/schema/{schema_name}",
+                        href=f"/stg/{repository.name}/schema/{schema_name}",
                         className="hover:underline"
                     ),
                     html.Span(" > "),
@@ -71,16 +72,18 @@ def layout(stage_name=None, schema_name=None, table_name=None):
         ],
         className="p-2"
         )
+    return layout
 
 
-@callback(
+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):
+    def activate_editable(n_clicks, df_src):
         if n_clicks is None:
             raise PreventUpdate
         if n_clicks > 0:
@@ -88,39 +91,40 @@ def activate_editable(n_clicks, df_src):
             return True, df_backup
         raise PreventUpdate
 
-@callback(
+    @app.callback(
-    Output("datatable", 'editable'),
+        Output("datatable", 'editable', allow_duplicate=True),
-    Output("datatable", 'data'),
+        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):
+    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
 
-# @callback(
+    @app.callback(
-#     Output("datatable", 'editable'),
+        Output("datatable", 'editable'),
-#     Input("btn_save", "n_clicks"),
+        Output("datatable", 'data'),
-#     State("datatable", 'editable'),
+        Input("btn_save", "n_clicks"),
-# )
+        State("datatable", 'editable'),
-# def save_modifications(n_clicks, editable):
+    )
-#     if n_clicks is None:
+    def save_modifications(n_clicks, editable):
-#         raise PreventUpdate
+        if n_clicks is None:
-#     if n_clicks > 0:
+            raise PreventUpdate
-#         return not editable
+        if n_clicks > 0:
-#     return editable
+            return not editable
+        return editable
 
-@callback(
+    @app.callback(
         Output("btn_edit", "style"),
         Output("btn_save", "style"),
         Output("btn_cancel", "style"),
         Input("datatable", "editable"),
-)
+    )
-def toolbar(editable):
+    def toolbar(editable):
         if editable:
             return {"display": "none"}, {"display": "block"}, {"display": "block"}
         return {"display": "block"}, {"display": "none"}, {"display": "none"}

requirements.txt

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

tests/repository/fs_examples/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/repository/test_fs_repository.py

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

@@ -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]}))