Compare commits
No commits in common. "dashboard" and "main" have entirely different histories.
@ -1,64 +0,0 @@
|
|||||||
import dash
|
|
||||||
from dash import Dash, dcc, html
|
|
||||||
|
|
||||||
from .datalake import stages
|
|
||||||
from .pages import config, home, repository, schema, table
|
|
||||||
|
|
||||||
external_scripts = [{"src": "https://cdn.tailwindcss.com"}]
|
|
||||||
# external_script = ["https://tailwindcss.com/", {"src": "https://cdn.tailwindcss.com"}]
|
|
||||||
|
|
||||||
app = Dash(
|
|
||||||
__name__,
|
|
||||||
use_pages=True,
|
|
||||||
external_scripts=external_scripts,
|
|
||||||
suppress_callback_exceptions=True,
|
|
||||||
)
|
|
||||||
app.scripts.config.serve_locally = True
|
|
||||||
dash.register_page(
|
|
||||||
home.__name__,
|
|
||||||
path="/",
|
|
||||||
layout=home.layout,
|
|
||||||
)
|
|
||||||
dash.register_page(config.__name__, path="/config", layout=config.layout)
|
|
||||||
dash.register_page(
|
|
||||||
repository.__name__,
|
|
||||||
path_template="/repository/<repository_name>",
|
|
||||||
layout=repository.layout_factory(stages),
|
|
||||||
)
|
|
||||||
dash.register_page(
|
|
||||||
schema.__name__,
|
|
||||||
path_template="/stg/<repository_name>/schema/<schema_name>",
|
|
||||||
layout=schema.layout_factory(stages),
|
|
||||||
)
|
|
||||||
dash.register_page(
|
|
||||||
table.__name__,
|
|
||||||
path_template="/stg/<repository_name>/schm/<schema_name>/table/<table_name>",
|
|
||||||
layout=table.layout_factory(stages),
|
|
||||||
)
|
|
||||||
table.callback_factory(app)
|
|
||||||
|
|
||||||
app.layout = html.Div(
|
|
||||||
[
|
|
||||||
html.Div(
|
|
||||||
[
|
|
||||||
dcc.Link(
|
|
||||||
html.H1(
|
|
||||||
"Plesna",
|
|
||||||
),
|
|
||||||
href="/",
|
|
||||||
className="text-4xl p-4 text-center grow align-baseline",
|
|
||||||
),
|
|
||||||
dcc.Link(
|
|
||||||
"Config",
|
|
||||||
href="/config",
|
|
||||||
className="flex-none hover:bg-amber-100 p-4 align-middle",
|
|
||||||
),
|
|
||||||
],
|
|
||||||
className="bg-amber-300 flex flex-row shadow",
|
|
||||||
),
|
|
||||||
dash.page_container,
|
|
||||||
]
|
|
||||||
)
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
app.run(debug=True)
|
|
||||||
@ -1,57 +0,0 @@
|
|||||||
from dash import dcc, html
|
|
||||||
|
|
||||||
from ..libs.repository.repository import AbstractRepository
|
|
||||||
|
|
||||||
|
|
||||||
def html_list_schema(stage:AbstractRepository, with_tables=True):
|
|
||||||
""" Build html list of schema in stage """
|
|
||||||
ul_classes = "ml-2"
|
|
||||||
schema_baseurl = f"/stg/{stage.name}/schema/"
|
|
||||||
if with_tables:
|
|
||||||
return html.Ul(
|
|
||||||
[
|
|
||||||
html.Li(
|
|
||||||
children = [
|
|
||||||
dcc.Link(
|
|
||||||
schema,
|
|
||||||
href=schema_baseurl + schema,
|
|
||||||
className="text-lg hover:underline"
|
|
||||||
),
|
|
||||||
html_list_table(stage, schema)
|
|
||||||
],
|
|
||||||
className=""
|
|
||||||
) for schema in stage.schemas()
|
|
||||||
],
|
|
||||||
className=ul_classes
|
|
||||||
)
|
|
||||||
return html.Ul(
|
|
||||||
[
|
|
||||||
html.Li(
|
|
||||||
dcc.Link(
|
|
||||||
schema,
|
|
||||||
href=schema_baseurl + schema,
|
|
||||||
className="text-lg hover:underline"
|
|
||||||
),
|
|
||||||
) for schema in stage.schemas()
|
|
||||||
],
|
|
||||||
className=ul_classes
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def html_list_table(stage:AbstractRepository, schema:str):
|
|
||||||
""" Build html list of table in stage """
|
|
||||||
table_baseurl = f"/stg/{stage.name}/schm/{schema}/table/"
|
|
||||||
return html.Ul(
|
|
||||||
[
|
|
||||||
html.Li(
|
|
||||||
dcc.Link(
|
|
||||||
table,
|
|
||||||
href=table_baseurl + table,
|
|
||||||
className="hover:underline"
|
|
||||||
),
|
|
||||||
) for table in stage.tables(schema=schema)
|
|
||||||
],
|
|
||||||
className="ml-4"
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
@ -1,14 +0,0 @@
|
|||||||
from dotenv import dotenv_values
|
|
||||||
|
|
||||||
from .libs.repository.fs_repository import FSRepository
|
|
||||||
|
|
||||||
env = {
|
|
||||||
**dotenv_values(".env"),
|
|
||||||
}
|
|
||||||
|
|
||||||
stages = {
|
|
||||||
"raw": FSRepository("raw", f"{env['DATA_PATH']}/{env['RAW_SUBPATH']}"),
|
|
||||||
"staging": FSRepository("staging", f"{env['DATA_PATH']}/{env['STAGING_SUBPATH']}"),
|
|
||||||
"gold": FSRepository("gold", f"{env['DATA_PATH']}/{env['GOLD_SUBPATH']}"),
|
|
||||||
"mart": FSRepository("mart", f"{env['DATA_PATH']}/{env['MART_SUBPATH']}"),
|
|
||||||
}
|
|
||||||
@ -1,70 +0,0 @@
|
|||||||
from collections.abc import Callable
|
|
||||||
from datetime import datetime
|
|
||||||
|
|
||||||
import pandas as pd
|
|
||||||
from pydantic import BaseModel
|
|
||||||
|
|
||||||
from ..repository.repository import AbstractRepository
|
|
||||||
|
|
||||||
|
|
||||||
class Schema(BaseModel):
|
|
||||||
repository: str
|
|
||||||
schema: str
|
|
||||||
|
|
||||||
|
|
||||||
class Table(BaseModel):
|
|
||||||
repository: str
|
|
||||||
schema: str
|
|
||||||
table: str
|
|
||||||
|
|
||||||
|
|
||||||
class Flux(BaseModel):
|
|
||||||
sources: list[Table]
|
|
||||||
destinations: dict[str, Table]
|
|
||||||
transformation: Callable[[list[pd.DataFrame]], dict[str, pd.DataFrame]]
|
|
||||||
|
|
||||||
|
|
||||||
class State(BaseModel):
|
|
||||||
statuses: dict[str, dict]
|
|
||||||
qty_out: int
|
|
||||||
failed_lines: list[str]
|
|
||||||
start: datetime
|
|
||||||
end: datetime
|
|
||||||
|
|
||||||
|
|
||||||
Repositories = dict[str, AbstractRepository]
|
|
||||||
|
|
||||||
|
|
||||||
def open_source(repositories: Repositories, source: Table) -> pd.DataFrame:
|
|
||||||
return repositories[source.repository].read(source.table, source.schema)
|
|
||||||
|
|
||||||
|
|
||||||
def write_source(
|
|
||||||
content: pd.DataFrame, repositories: Repositories, destination: Table
|
|
||||||
) -> str:
|
|
||||||
return repositories[destination.repository].write(
|
|
||||||
content, destination.table, destination.schema
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def consume_flux(flux: Flux, repositories: dict[str, AbstractRepository]) -> State:
|
|
||||||
start = datetime.now()
|
|
||||||
src_dfs = [open_source(repositories, source) for source in flux.sources]
|
|
||||||
|
|
||||||
built_dfs = flux.transformation(src_dfs)
|
|
||||||
|
|
||||||
statuses = {
|
|
||||||
dest: write_source(df, repositories, flux.destinations[dest])
|
|
||||||
for dest, df in built_dfs.items()
|
|
||||||
}
|
|
||||||
|
|
||||||
end = datetime.now()
|
|
||||||
qty_out = 0
|
|
||||||
failed_lines = []
|
|
||||||
return State(
|
|
||||||
statuses=statuses,
|
|
||||||
qty_out=qty_out,
|
|
||||||
failed_lines=failed_lines,
|
|
||||||
start=start,
|
|
||||||
end=end,
|
|
||||||
)
|
|
||||||
@ -1,86 +0,0 @@
|
|||||||
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
|
|
||||||
@ -1,5 +0,0 @@
|
|||||||
from abc import ABC
|
|
||||||
|
|
||||||
|
|
||||||
class AbstractMetadataEngine(ABC):
|
|
||||||
pass
|
|
||||||
@ -1,37 +0,0 @@
|
|||||||
import abc
|
|
||||||
|
|
||||||
from .metadata import AbstractMetadataEngine
|
|
||||||
|
|
||||||
|
|
||||||
class AbstractRepository(abc.ABC):
|
|
||||||
metadata_engine = AbstractMetadataEngine
|
|
||||||
|
|
||||||
@abc.abstractmethod
|
|
||||||
def schemas(self) -> list[str]:
|
|
||||||
"""List schemas"""
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
@abc.abstractmethod
|
|
||||||
def tables(self, schema) -> list[str]:
|
|
||||||
"""List table in schema"""
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
@abc.abstractmethod
|
|
||||||
def infos(self, table: str, schema: str) -> dict[str, str]:
|
|
||||||
"""Get infos about the table"""
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
@abc.abstractmethod
|
|
||||||
def read(self, table: str, schema: str):
|
|
||||||
"""Get content of the table"""
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
@abc.abstractmethod
|
|
||||||
def write(self, content, table: str, schema: str):
|
|
||||||
"""Write content into the table"""
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
@abc.abstractmethod
|
|
||||||
def delete_table(self, table: str, schema: str):
|
|
||||||
"""Delete the table"""
|
|
||||||
raise NotImplementedError
|
|
||||||
@ -1,14 +0,0 @@
|
|||||||
from dash import html
|
|
||||||
from dotenv import dotenv_values
|
|
||||||
import os
|
|
||||||
|
|
||||||
env = {
|
|
||||||
**dotenv_values(".env"),
|
|
||||||
**os.environ,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
layout = html.Div([
|
|
||||||
html.H1('This is our Config page'),
|
|
||||||
html.Ul(children = [html.Li(f"{k} = {v}") for k,v in env.items()]),
|
|
||||||
])
|
|
||||||
@ -1,27 +0,0 @@
|
|||||||
from dash import dcc, html
|
|
||||||
|
|
||||||
from ..components.lists import html_list_schema
|
|
||||||
from ..datalake import stages
|
|
||||||
|
|
||||||
layout = html.Div([
|
|
||||||
html.Div(children=[
|
|
||||||
html.Ul(
|
|
||||||
children=[
|
|
||||||
html.Li(
|
|
||||||
children=[
|
|
||||||
dcc.Link(
|
|
||||||
stagename,
|
|
||||||
href=f"/stage/{stagename}",
|
|
||||||
className="text-2xl text-center p-2 bg-amber-100 rounded shadow"
|
|
||||||
),
|
|
||||||
html_list_schema(stage)
|
|
||||||
],
|
|
||||||
className="flex-1 bg-gray-100 rounded flex flex-col shadow"
|
|
||||||
) for stagename, stage in stages.items()
|
|
||||||
],
|
|
||||||
className="flex flex-row space-x-2"
|
|
||||||
)
|
|
||||||
],
|
|
||||||
className="w-full mt-4 px-2"
|
|
||||||
),
|
|
||||||
])
|
|
||||||
@ -1,18 +0,0 @@
|
|||||||
from dash import html
|
|
||||||
|
|
||||||
from ..components.lists import html_list_schema
|
|
||||||
from ..libs.repository.repository import AbstractRepository
|
|
||||||
|
|
||||||
|
|
||||||
def layout_factory(repositories: dict[str, AbstractRepository]):
|
|
||||||
def layout(repository_name: str = ""):
|
|
||||||
repository = repositories[repository_name]
|
|
||||||
return html.Div(
|
|
||||||
[
|
|
||||||
html.H2(f"{repository.name}", className="text-2xl p-4 py-2"),
|
|
||||||
html_list_schema(repository),
|
|
||||||
],
|
|
||||||
className="flex flex-col",
|
|
||||||
)
|
|
||||||
|
|
||||||
return layout
|
|
||||||
@ -1,28 +0,0 @@
|
|||||||
from dash import dcc, html
|
|
||||||
|
|
||||||
from ..libs.repository.repository import AbstractRepository
|
|
||||||
|
|
||||||
|
|
||||||
def layout_factory(repositories: dict[str, AbstractRepository]):
|
|
||||||
def layout(repository_name: str = "", schema_name: str = ""):
|
|
||||||
repository = repositories[repository_name]
|
|
||||||
return html.Div(
|
|
||||||
[
|
|
||||||
html.H2(
|
|
||||||
[
|
|
||||||
dcc.Link(
|
|
||||||
f"{repository.name}",
|
|
||||||
href=f"/repository/{repository.name}",
|
|
||||||
className="hover:underline",
|
|
||||||
),
|
|
||||||
html.Span(" > "),
|
|
||||||
html.Span(
|
|
||||||
f"{schema_name}",
|
|
||||||
),
|
|
||||||
],
|
|
||||||
className="text-2xl p-4 py-2",
|
|
||||||
),
|
|
||||||
]
|
|
||||||
)
|
|
||||||
|
|
||||||
return layout
|
|
||||||
@ -1,130 +0,0 @@
|
|||||||
from dash import Input, Output, State, dash_table, dcc, html
|
|
||||||
from dash.exceptions import PreventUpdate
|
|
||||||
|
|
||||||
from ..libs.repository.repository import AbstractRepository
|
|
||||||
|
|
||||||
|
|
||||||
def layout_factory(repositories: dict[str,AbstractRepository]):
|
|
||||||
def layout(repository_name:str="", schema_name:str="", table_name:str=""):
|
|
||||||
repository = repositories[repository_name]
|
|
||||||
df = repository.read(table=table_name, schema=schema_name)
|
|
||||||
return html.Div([
|
|
||||||
dcc.Store(id="table_backup"),
|
|
||||||
html.Div([
|
|
||||||
html.H2([
|
|
||||||
dcc.Link(
|
|
||||||
f"{repository.name}",
|
|
||||||
href=f"/repository/{repository.name}",
|
|
||||||
className="hover:underline"
|
|
||||||
),
|
|
||||||
html.Span(" > "),
|
|
||||||
dcc.Link(
|
|
||||||
f"{schema_name}",
|
|
||||||
href=f"/stg/{repository.name}/schema/{schema_name}",
|
|
||||||
className="hover:underline"
|
|
||||||
),
|
|
||||||
html.Span(" > "),
|
|
||||||
html.Span(table_name),
|
|
||||||
],
|
|
||||||
className="text-2xl"
|
|
||||||
),
|
|
||||||
html.Div([
|
|
||||||
html.Button(
|
|
||||||
"Editer",
|
|
||||||
id="btn_edit",
|
|
||||||
className="rounded border px-2 py-1",
|
|
||||||
style={"display": "block"}
|
|
||||||
),
|
|
||||||
html.Button(
|
|
||||||
"Sauver",
|
|
||||||
id="btn_save",
|
|
||||||
className="rounded border px-2 py-1 border-green-500 hover:bg-green-500",
|
|
||||||
style={"display": "none"}
|
|
||||||
),
|
|
||||||
html.Button(
|
|
||||||
"Annuler",
|
|
||||||
id="btn_cancel",
|
|
||||||
className="rounded border px-2 py-1 border-red-500 hover:bg-red-500",
|
|
||||||
style={"display": "none"}
|
|
||||||
),
|
|
||||||
],
|
|
||||||
className="flex flex-row space-x-2",
|
|
||||||
id="toolbar"
|
|
||||||
),
|
|
||||||
],
|
|
||||||
className="flex flex-row justify-between p-4"
|
|
||||||
),
|
|
||||||
html.Div([
|
|
||||||
html.Div([
|
|
||||||
dash_table.DataTable(
|
|
||||||
id="datatable",
|
|
||||||
data=df.to_dict('records'),
|
|
||||||
columns=[{"name": i, "id": i} for i in df.columns],
|
|
||||||
filter_action="native",
|
|
||||||
sort_action="native",
|
|
||||||
sort_mode="multi",
|
|
||||||
editable=False
|
|
||||||
)
|
|
||||||
])
|
|
||||||
],
|
|
||||||
className="overflow-y-auto"
|
|
||||||
),
|
|
||||||
],
|
|
||||||
className="p-2"
|
|
||||||
)
|
|
||||||
return layout
|
|
||||||
|
|
||||||
|
|
||||||
def callback_factory(app):
|
|
||||||
@app.callback(
|
|
||||||
Output("datatable", 'editable', allow_duplicate=True),
|
|
||||||
Output("table_backup", 'data'),
|
|
||||||
Input("btn_edit", "n_clicks"),
|
|
||||||
State("datatable", 'data'),
|
|
||||||
prevent_initial_call=True
|
|
||||||
)
|
|
||||||
def activate_editable(n_clicks, df_src):
|
|
||||||
if n_clicks is None:
|
|
||||||
raise PreventUpdate
|
|
||||||
if n_clicks > 0:
|
|
||||||
df_backup = df_src.copy()
|
|
||||||
return True, df_backup
|
|
||||||
raise PreventUpdate
|
|
||||||
|
|
||||||
@app.callback(
|
|
||||||
Output("datatable", 'editable', allow_duplicate=True),
|
|
||||||
Output("datatable", 'data', allow_duplicate=True),
|
|
||||||
Input("btn_cancel", "n_clicks"),
|
|
||||||
State("table_backup", 'data'),
|
|
||||||
prevent_initial_call=True
|
|
||||||
)
|
|
||||||
def cancel_modifications(n_clicks, data):
|
|
||||||
if n_clicks is None:
|
|
||||||
raise PreventUpdate
|
|
||||||
if n_clicks > 0 and data is not None:
|
|
||||||
return False, data.copy()
|
|
||||||
raise PreventUpdate
|
|
||||||
|
|
||||||
@app.callback(
|
|
||||||
Output("datatable", 'editable'),
|
|
||||||
Output("datatable", 'data'),
|
|
||||||
Input("btn_save", "n_clicks"),
|
|
||||||
State("datatable", 'editable'),
|
|
||||||
)
|
|
||||||
def save_modifications(n_clicks, editable):
|
|
||||||
if n_clicks is None:
|
|
||||||
raise PreventUpdate
|
|
||||||
if n_clicks > 0:
|
|
||||||
return not editable
|
|
||||||
return editable
|
|
||||||
|
|
||||||
@app.callback(
|
|
||||||
Output("btn_edit", "style"),
|
|
||||||
Output("btn_save", "style"),
|
|
||||||
Output("btn_cancel", "style"),
|
|
||||||
Input("datatable", "editable"),
|
|
||||||
)
|
|
||||||
def toolbar(editable):
|
|
||||||
if editable:
|
|
||||||
return {"display": "none"}, {"display": "block"}, {"display": "block"}
|
|
||||||
return {"display": "block"}, {"display": "none"}, {"display": "none"}
|
|
||||||
@ -1,6 +1,7 @@
|
|||||||
jupyter==1.0.0
|
jupyter==1.0.0
|
||||||
pandas==2.2.2
|
pandas==1.5.0
|
||||||
pydantic==2.8.2
|
pdf-oralia==0.3.11
|
||||||
|
pydantic==2.6.1
|
||||||
click==8.1.7
|
click==8.1.7
|
||||||
openpyxl==3.1.5
|
dlt[duckdb]>=0.4.3a0
|
||||||
xlrd==2.0.1
|
openpyxl>=3.0.0
|
||||||
|
|||||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
@ -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
|
|
||||||
|
|
||||||
|
@ -1,84 +0,0 @@
|
|||||||
import shutil
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
from pandas import pandas
|
|
||||||
|
|
||||||
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",
|
|
||||||
"username-password-recovery-code.xls",
|
|
||||||
]
|
|
||||||
assert repo.tables("salary") == ["salary.pdf"]
|
|
||||||
|
|
||||||
|
|
||||||
def test_read_csv(location):
|
|
||||||
repo = FSRepository("example", location)
|
|
||||||
username = repo.read("username.csv", "username", delimiter=";")
|
|
||||||
assert list(username.columns) == [
|
|
||||||
"Username",
|
|
||||||
"Identifier",
|
|
||||||
"First name",
|
|
||||||
"Last name",
|
|
||||||
]
|
|
||||||
assert len(username.index) == 5
|
|
||||||
|
|
||||||
|
|
||||||
def test_fake_read_xlsx(location):
|
|
||||||
repo = FSRepository("example", location)
|
|
||||||
df = pandas.read_excel(
|
|
||||||
location / "username" / "username-password-recovery-code.xls"
|
|
||||||
)
|
|
||||||
print(df)
|
|
||||||
|
|
||||||
|
|
||||||
def test_read_xlsx(location):
|
|
||||||
repo = FSRepository("example", location)
|
|
||||||
username = repo.read("username-password-recovery-code.xls", "username")
|
|
||||||
assert list(username.columns) == [
|
|
||||||
"Username",
|
|
||||||
"Identifier",
|
|
||||||
"One-time password",
|
|
||||||
"Recovery code",
|
|
||||||
"First name",
|
|
||||||
"Last name",
|
|
||||||
"Department",
|
|
||||||
"Location",
|
|
||||||
]
|
|
||||||
assert len(username.index) == 5
|
|
||||||
@ -1,131 +0,0 @@
|
|||||||
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
|
|
||||||
FakeSchema = dict[str, pd.DataFrame]
|
|
||||||
FakeSchemas = dict[str, FakeSchema]
|
|
||||||
|
|
||||||
|
|
||||||
class FakeRepository(AbstractRepository):
|
|
||||||
def __init__(self, schemas: FakeSchemas):
|
|
||||||
self._schemas = {}
|
|
||||||
for schema_name, tables in schemas.items():
|
|
||||||
schema = {}
|
|
||||||
for table, df in tables.items():
|
|
||||||
schema[table] = {
|
|
||||||
"df": df,
|
|
||||||
"metadata": {
|
|
||||||
"status": "new",
|
|
||||||
"qty_read": 0,
|
|
||||||
"qty_write": 0,
|
|
||||||
},
|
|
||||||
}
|
|
||||||
self._schemas[schema_name] = schema
|
|
||||||
|
|
||||||
def schemas(self):
|
|
||||||
"""List schemas"""
|
|
||||||
return list(self._schemas.keys())
|
|
||||||
|
|
||||||
def tables(self, schema):
|
|
||||||
"""List table's name in schema"""
|
|
||||||
return list(self._schemas[schema].keys())
|
|
||||||
|
|
||||||
def infos(self, table: str, schema: str) -> dict[str, str]:
|
|
||||||
"""Get infos about the table"""
|
|
||||||
return self._schemas[schema][table]["metadata"]
|
|
||||||
|
|
||||||
def read(self, table, schema) -> pd.DataFrame:
|
|
||||||
"""Get content of the table"""
|
|
||||||
self._schemas[schema][table]["metadata"]["qty_read"] += 1
|
|
||||||
return self._schemas[schema][table]["df"]
|
|
||||||
|
|
||||||
def write(self, content, table, schema) -> dict[str, str]:
|
|
||||||
"""Write content into the table"""
|
|
||||||
try:
|
|
||||||
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)
|
|
||||||
|
|
||||||
def delete_table(self, table, schema):
|
|
||||||
"""Delete the table"""
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
|
|
||||||
def test_fakerepository():
|
|
||||||
fakerepository = FakeRepository(
|
|
||||||
{
|
|
||||||
"foo": {
|
|
||||||
"table1": pd.DataFrame({"A": []}),
|
|
||||||
"table2": pd.DataFrame({"B": []}),
|
|
||||||
},
|
|
||||||
"bar": {
|
|
||||||
"table1": pd.DataFrame({"C": []}),
|
|
||||||
"table2": pd.DataFrame({"D": []}),
|
|
||||||
},
|
|
||||||
}
|
|
||||||
)
|
|
||||||
assert fakerepository.schemas() == ["foo", "bar"]
|
|
||||||
assert fakerepository.tables("foo") == ["table1", "table2"]
|
|
||||||
assert fakerepository.infos("table1", "foo") == {
|
|
||||||
"status": "new",
|
|
||||||
"qty_read": 0,
|
|
||||||
"qty_write": 0,
|
|
||||||
}
|
|
||||||
assert fakerepository.read("table1", "foo").equals(pd.DataFrame({"A": []}))
|
|
||||||
assert fakerepository.infos("table1", "foo") == {
|
|
||||||
"status": "new",
|
|
||||||
"qty_read": 1,
|
|
||||||
"qty_write": 0,
|
|
||||||
}
|
|
||||||
|
|
||||||
df = pd.DataFrame({"A": [1, 2]})
|
|
||||||
assert fakerepository.write(df, "table1", "foo") == {
|
|
||||||
"status": "modified",
|
|
||||||
"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]}))
|
|
||||||
Loading…
Reference in New Issue
Block a user