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

Feat: test on pandas xlsx and ods file reader

Browse Source
This commit is contained in:
Bertrand Benjamin
2024-10-07 05:27:46 +02:00
parent 5dfc1c9751
commit c347deee85
6 changed files with 69 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -4,6 +4,10 @@ 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):
@@ -20,18 +24,30 @@ class FSRepository(AbstractRepository):
if only_files:
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() and not str(f).startswith(".")
]
if only_directories:
if recursive:
return [str(f[0].relative_to(dirpath)) for f in dirpath.walk()]
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()
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()]
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)
@@ -49,16 +65,17 @@ class FSRepository(AbstractRepository):
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)
assert table_path.exists()
extension = table_path.suffix
if extension == ".csv":
if extension in ACCEPTABLE_EXTENTIONS["csv"]:
return pd.read_csv(table_path, **read_options)
if extension == ".xlsx":
return pd.read_excel(table_path, **read_options)
if extension in ACCEPTABLE_EXTENTIONS["excel"]:
return pd.read_excel(table_path, engine = "openpyxl", **read_options)
raise ValueError("Can't open the table")
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)