Compare commits
No commits in common. "9d45625a5ed017fba77cf77a37e137cca0b9578f" and "226ce84dcea7e0c400d808ec559e386f2f93c82e" have entirely different histories.
9d45625a5e
...
226ce84dce
@ -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,8 +0,0 @@
|
|||||||
from plesna.models.flux import Flux, FluxMetaData
|
|
||||||
|
|
||||||
|
|
||||||
def consume_flux(flux: Flux) -> FluxMetaData:
|
|
||||||
metadata = flux.transformation.function(
|
|
||||||
sources=flux.sources, targets=flux.targets, **flux.transformation.extra_kwrds
|
|
||||||
)
|
|
||||||
return FluxMetaData(data=metadata)
|
|
||||||
@ -1,21 +0,0 @@
|
|||||||
import abc
|
|
||||||
|
|
||||||
|
|
||||||
class DataCatalogue:
|
|
||||||
def __init__(self):
|
|
||||||
pass
|
|
||||||
|
|
||||||
@abc.abstractmethod
|
|
||||||
def schemas(self) -> dict[str:str]:
|
|
||||||
"""List schemas"""
|
|
||||||
raise NotImplementedError
|
|
||||||
|
|
||||||
@abc.abstractmethod
|
|
||||||
def tables(self, schema) -> dict[str: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
|
|
||||||
@ -1,3 +0,0 @@
|
|||||||
class DataStore:
|
|
||||||
def __init__(self, name):
|
|
||||||
self._name
|
|
||||||
@ -1,83 +0,0 @@
|
|||||||
from pathlib import Path
|
|
||||||
|
|
||||||
from pydantic import BaseModel, computed_field
|
|
||||||
|
|
||||||
from plesna.models.storage import Schema, Table
|
|
||||||
|
|
||||||
from .datacatalogue import DataCatalogue
|
|
||||||
|
|
||||||
|
|
||||||
class FSSchema(BaseModel):
|
|
||||||
path: Path
|
|
||||||
|
|
||||||
@computed_field
|
|
||||||
@property
|
|
||||||
def ref(self) -> Schema:
|
|
||||||
return Schema(
|
|
||||||
id=str(self.path),
|
|
||||||
value=str(self.path),
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class FSTable(BaseModel):
|
|
||||||
path: Path
|
|
||||||
|
|
||||||
@computed_field
|
|
||||||
@property
|
|
||||||
def ref(self) -> Table:
|
|
||||||
return Table(
|
|
||||||
id=str(self.path),
|
|
||||||
value=str(self.path),
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class FSDataCatalogue(DataCatalogue):
|
|
||||||
"""DataCatalogue based on files tree structure"""
|
|
||||||
|
|
||||||
def __init__(self, name: str, basepath: str = "."):
|
|
||||||
self._basepath = Path(basepath)
|
|
||||||
self.name = name
|
|
||||||
|
|
||||||
assert self._basepath.exists()
|
|
||||||
|
|
||||||
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)}
|
|
||||||
@ -31,6 +31,3 @@ class GraphSet:
|
|||||||
@property
|
@property
|
||||||
def node_sets(self):
|
def node_sets(self):
|
||||||
return self._node_sets
|
return self._node_sets
|
||||||
|
|
||||||
def is_valid_dag(self):
|
|
||||||
pass
|
|
||||||
@ -1,14 +0,0 @@
|
|||||||
from pydantic import BaseModel
|
|
||||||
|
|
||||||
from plesna.models.storage import Table
|
|
||||||
from plesna.models.transformation import Transformation
|
|
||||||
|
|
||||||
|
|
||||||
class Flux(BaseModel):
|
|
||||||
sources: dict[str, Table]
|
|
||||||
targets: dict[str, Table]
|
|
||||||
transformation: Transformation
|
|
||||||
|
|
||||||
|
|
||||||
class FluxMetaData(BaseModel):
|
|
||||||
data: dict
|
|
||||||
@ -1,25 +0,0 @@
|
|||||||
from pydantic import BaseModel
|
|
||||||
|
|
||||||
|
|
||||||
class Schema(BaseModel):
|
|
||||||
"""Logical agregation for Table
|
|
||||||
|
|
||||||
id: uniq identifier for the schema
|
|
||||||
value: string which describe where to find the schema in the storage system
|
|
||||||
|
|
||||||
"""
|
|
||||||
|
|
||||||
id: str
|
|
||||||
value: str
|
|
||||||
|
|
||||||
|
|
||||||
class Table(BaseModel):
|
|
||||||
"""Place where data are stored
|
|
||||||
|
|
||||||
id: uniq identifier for the table
|
|
||||||
value: string which describe where to find the table in the storage system
|
|
||||||
|
|
||||||
"""
|
|
||||||
|
|
||||||
id: str
|
|
||||||
value: str
|
|
||||||
@ -1,15 +0,0 @@
|
|||||||
from collections.abc import Callable
|
|
||||||
|
|
||||||
from pydantic import BaseModel
|
|
||||||
|
|
||||||
|
|
||||||
class Transformation(BaseModel):
|
|
||||||
"""
|
|
||||||
The function have to have at least 2 arguments: sources and targets
|
|
||||||
Other arguments will came throught extra_kwrds
|
|
||||||
|
|
||||||
The function will have to return metadata as dict
|
|
||||||
"""
|
|
||||||
|
|
||||||
function: Callable
|
|
||||||
extra_kwrds: dict = {}
|
|
||||||
@ -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
|
||||||
|
|||||||
@ -1,35 +0,0 @@
|
|||||||
from plesna.compute.consume_flux import consume_flux
|
|
||||||
from plesna.models.flux import Flux
|
|
||||||
from plesna.models.storage import Table
|
|
||||||
from plesna.models.transformation import Transformation
|
|
||||||
|
|
||||||
|
|
||||||
def test_consume_flux():
|
|
||||||
sources = {
|
|
||||||
"src1": Table(id="src1", value="here"),
|
|
||||||
"src2": Table(id="src2", value="here"),
|
|
||||||
}
|
|
||||||
targets = {
|
|
||||||
"tgt1": Table(id="tgt1", value="this"),
|
|
||||||
"tgt2": Table(id="tgt2", value="that"),
|
|
||||||
}
|
|
||||||
|
|
||||||
def func(sources, targets, **kwrds):
|
|
||||||
return {
|
|
||||||
"sources": len(sources),
|
|
||||||
"targets": len(targets),
|
|
||||||
"kwrds": len(kwrds),
|
|
||||||
}
|
|
||||||
|
|
||||||
flux = Flux(
|
|
||||||
sources=sources,
|
|
||||||
targets=targets,
|
|
||||||
transformation=Transformation(function=func, extra_kwrds={"extra": "super"}),
|
|
||||||
)
|
|
||||||
|
|
||||||
meta = consume_flux(flux)
|
|
||||||
assert meta.data == {
|
|
||||||
"sources": 2,
|
|
||||||
"targets": 2,
|
|
||||||
"kwrds": 1,
|
|
||||||
}
|
|
||||||
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,72 +0,0 @@
|
|||||||
import shutil
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
|
|
||||||
from plesna.datastore.fs_datacatalogue import FSDataCatalogue
|
|
||||||
|
|
||||||
FIXTURE_DIR = Path(__file__).parent / Path("./fs_files/")
|
|
||||||
|
|
||||||
|
|
||||||
@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 = FIXTURE_DIR
|
|
||||||
assert example_src.exists()
|
|
||||||
|
|
||||||
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 = FSDataCatalogue("example", location)
|
|
||||||
assert repo.ls() == [
|
|
||||||
"username",
|
|
||||||
"salary",
|
|
||||||
]
|
|
||||||
|
|
||||||
assert repo.ls(recursive=True) == [
|
|
||||||
"username",
|
|
||||||
"salary",
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
def test_list_schema(location):
|
|
||||||
repo = FSDataCatalogue("example", location)
|
|
||||||
assert {id: s.model_dump()["ref"]["id"] for id, s in repo.schemas().items()} == {
|
|
||||||
".": ".",
|
|
||||||
"username": "username",
|
|
||||||
"salary": "salary",
|
|
||||||
}
|
|
||||||
assert {id: s.model_dump()["ref"]["value"] for id, s in repo.schemas().items()} == {
|
|
||||||
".": ".",
|
|
||||||
"username": "username",
|
|
||||||
"salary": "salary",
|
|
||||||
}
|
|
||||||
assert {id: s.model_dump()["path"] for id, s in repo.schemas().items()} == {
|
|
||||||
".": Path("."),
|
|
||||||
"username": Path("username"),
|
|
||||||
"salary": Path("salary"),
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def test_list_tables(location):
|
|
||||||
repo = FSDataCatalogue("example", location)
|
|
||||||
assert repo.tables() == {}
|
|
||||||
assert {id: t.model_dump()["ref"]["value"] for id,t in repo.tables("username").items()} == {
|
|
||||||
"username.csv": "username.csv",
|
|
||||||
"username-password-recovery-code.xlsx": "username-password-recovery-code.xlsx",
|
|
||||||
"username-password-recovery-code.xls": "username-password-recovery-code.xls",
|
|
||||||
}
|
|
||||||
assert {id: t.model_dump()["ref"]["value"] for id,t in repo.tables("salary").items()} == {
|
|
||||||
"salary.pdf": "salary.pdf",
|
|
||||||
}
|
|
||||||
@ -1,6 +1,6 @@
|
|||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
from plesna.graph.graph import Edge, Graph, Node
|
from plesna.graph import Edge, Graph, Node
|
||||||
|
|
||||||
|
|
||||||
def test_append_nodess():
|
def test_append_nodess():
|
||||||
|
|||||||
@ -1,4 +1,4 @@
|
|||||||
from plesna.graph.graph_set import EdgeOnSet, GraphSet, Node
|
from plesna.graph_set import EdgeOnSet, GraphSet, Node
|
||||||
|
|
||||||
|
|
||||||
def test_init():
|
def test_init():
|
||||||
|
|||||||
@ -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