plesna/dashboard/libs/repository/fs_repository.py

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