dashboard/libs/flux/flux.py

@@ -25,7 +25,7 @@ class Flux(BaseModel):
 
 
 class State(BaseModel):
-    statuses: dict[str, dict]
+    statuses: dict[str, str]
     qty_out: int
     failed_lines: list[str]
     start: datetime

dashboard/libs/repository/fs_repository.py

@@ -10,60 +10,55 @@ class FSRepository(AbstractRepository):
         self.name = name
 
         self.basepath = Path(basepath)
-        assert self.basepath.exists()
         self._metadata_engine = metadata_engine
 
-    def ls(
+    def ls(self, dir, only_files=False, only_directories=False, recursive=False) -> list[str]:
-        self, dir="", only_files=False, only_directories=False, recursive=False
+        dirpath = Path(dir)
-    ) -> list[str]:
-        dirpath = self.basepath / dir
 
         if only_files:
-            return [
+            return [str(f.relative_to(dirpath)) for f in dirpath.iterdir() if not f.is_dir()]
-                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 [
+            return [str(f.relative_to(dirpath)) for f in dirpath.iterdir() if f.is_dir()]
-                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]:
-        return self.ls("", only_directories=True, recursive=True)
+        dirpath = self.basepath
+        return self.ls(dirpath, only_directories=True, recursive=True)
 
-    def tables(self, schema: str = ".") -> list[str]:
+    def tables(self, schema:str) -> list[str]:
-        return self.ls(schema, only_files=True)
+        dirpath = self.basepath / schema
+        return self.ls(dirpath, only_files=True)
 
-    def build_table_path(self, table: str, schema: str):
+    def build_table_path(self, table:str, schema:str):
         table_path = self.basepath
-        if schema == ".":
+        if schema == '.':
             return table_path / table
         return table_path / schema / table
 
-    def infos(self, table: str, schema: str = "."):
+    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={}):
+    def read(self, table:str, schema:str='.', read_options={}):
         table_path = self.build_table_path(table, schema)
         extension = table_path.suffix
-        if extension == ".csv":
+        if extension == '.csv':
             return pd.read_csv(table_path, **read_options)
 
-        if extension == ".xlsx":
+        if extension == '.xlsx':
             return pd.read_excel(table_path, **read_options)
 
         raise ValueError("Can't open the table")
 
-    def write(self, content, table: str, schema: str = "."):
+    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 = "."):
+    def delete(self, table:str, schema:str='.'):
         table_path = self.build_table_path(table, schema)
         pass

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,46 +0,0 @@
-import shutil
-from pathlib import Path
-
-import pytest
-
-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",
-    ]
-    assert repo.tables("salary") == ["salary.pdf"]

tests/test_flux.py

@@ -1,7 +1,6 @@
 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
@@ -44,17 +43,7 @@ class FakeRepository(AbstractRepository):
 
     def write(self, content, table, schema) -> dict[str, str]:
         """Write content into the table"""
-        try:
+        self._schemas[schema][table]["df"] = content
-            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)
@@ -97,35 +86,3 @@ def test_fakerepository():
         "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]}))