lafrite/plesna
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Feat: start testing fs_repository

Browse Source
This commit is contained in:
Bertrand Benjamin
2024-08-14 10:44:38 +02:00
parent 7fc10128da
commit 5dfc1c9751
6 changed files with 75 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
dashboard/libs/repository/fs_repository.py
View File

@@ -10,55 +10,60 @@ class FSRepository(AbstractRepository):
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 = Path(dir)
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()]
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() 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)
return self.ls("", only_directories=True, recursive=True)
def tables(self, schema:str) -> list[str]:
dirpath = self.basepath / schema
return self.ls(dirpath, only_files=True)
def tables(self, schema: str = ".") -> list[str]:
return self.ls(schema, 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 info(self, table:str, schema:str='.'):
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={}):
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(self, table:str, schema:str='.'):
def delete_table(self, table: str, schema: str = "."):
table_path = self.build_table_path(table, schema)
pass