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
|
||||
pandas==2.2.2
|
||||
pydantic==2.8.2
|
||||
pandas==1.5.0
|
||||
pdf-oralia==0.3.11
|
||||
pydantic==2.6.1
|
||||
click==8.1.7
|
||||
openpyxl==3.1.5
|
||||
xlrd==2.0.1
|
||||
dlt[duckdb]>=0.4.3a0
|
||||
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