feat: start datacatalogue

plesna/dataplatform.py

@@ -0,0 +1,27 @@
+from plesna.datastore.datacatalogue import DataCatalogue
+from plesna.graph.graph_set import GraphSet
+
+
+class DataPlateformError(Exception):
+    pass
+
+
+class DataPlateform:
+    def __init__(self):
+        self._graphset = GraphSet()
+        self._metadata_engine = ""
+        self._transformations = {}
+        self._datacatalogues = {}
+
+    def add_datacatalague(self, name: str, datacatalogue: DataCatalogue):
+        if name in self._datacatalogues:
+            raise DataPlateformError("The datacatalogue {name} already exists")
+
+        self._datacatalogues[name] = datacatalogue
+
+    @property
+    def datacatalogues(self):
+        return list(self._datacatalogues)
+
+    def get_datacatalogue(self, name: str):
+        return self._datacatalogues[name]

plesna/datastore/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)}

tests/dataplatform/test_dataplateform.py

@@ -0,0 +1,43 @@
+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"
+    raw_path.mkdir()
+    return FSDataCatalogue("raw", raw_path)
+
+
+@pytest.fixture
+def bronze_catalogue(tmp_path):
+    bronze_path = Path(tmp_path) / "bronze"
+    bronze_path.mkdir()
+    return FSDataCatalogue("bronze", bronze_path)
+
+
+@pytest.fixture
+def silver_catalogue(tmp_path):
+    silver_path = Path(tmp_path) / "silver"
+    silver_path.mkdir()
+    return FSDataCatalogue("silver", silver_path)
+
+
+def test_add_catalogue(
+    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)
+
+    assert dp.datacatalogues == ["raw", "bronze", "silver"]
+    assert dp.get_datacatalogue("raw") == raw_catalogue

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