From 60527645f249c77482a2fd17acb3c101ae3af024 Mon Sep 17 00:00:00 2001 From: buenosairesam Date: Tue, 22 Sep 2026 08:18:25 -0300 Subject: [PATCH] distilled tools --- .../station/tools/dataconvert/.env-example | 17 +++++ soleprint/station/tools/dataconvert/README.md | 2 +- soleprint/station/tools/dataconvert/config.py | 34 +++++++-- .../dataconvert/dataconvert-example.json | 2 +- .../station/tools/dataconvert/dataconvert.py | 76 +++++++++++++------ soleprint/station/tools/dataconvert/output.py | 33 +++++--- soleprint/station/tools/dataconvert/sqlgen.py | 33 +++++++- soleprint/station/tools/dataconvert/uv.lock | 34 ++++----- .../station/tools/datasource/.env-example | 20 +++++ soleprint/station/tools/datasource/.gitignore | 4 + soleprint/station/tools/datasource/README.md | 26 +++++++ .../tools/datasource/casting_helpers.sql | 30 ++++++++ .../tools/datasource/dump_trasformed.sh | 20 +++++ .../tools/datasource/load_seed_data.sh | 59 ++++++++++++++ .../tools/datasource/migrate_schema.py | 54 +++++++++++++ .../tools/datasource/restore_transformed.sh | 18 +++++ .../datasource/schema_mapping-example.yaml | 21 +++++ 17 files changed, 422 insertions(+), 61 deletions(-) create mode 100644 soleprint/station/tools/dataconvert/.env-example create mode 100644 soleprint/station/tools/datasource/.env-example create mode 100644 soleprint/station/tools/datasource/.gitignore create mode 100644 soleprint/station/tools/datasource/README.md create mode 100644 soleprint/station/tools/datasource/casting_helpers.sql create mode 100644 soleprint/station/tools/datasource/dump_trasformed.sh create mode 100644 soleprint/station/tools/datasource/load_seed_data.sh create mode 100644 soleprint/station/tools/datasource/migrate_schema.py create mode 100644 soleprint/station/tools/datasource/restore_transformed.sh create mode 100644 soleprint/station/tools/datasource/schema_mapping-example.yaml diff --git a/soleprint/station/tools/dataconvert/.env-example b/soleprint/station/tools/dataconvert/.env-example new file mode 100644 index 0000000..5bcdec7 --- /dev/null +++ b/soleprint/station/tools/dataconvert/.env-example @@ -0,0 +1,17 @@ +# Environment Variables for dataconvert tool + +# Input settings +DATACONVERT_INPUT="data/*" +DATACONVERT_EXCLUDE="*.tmp" +DATACONVERT_CONFIG= + +# Output settings +DATACONVERT_OUT_DIR=datadir/sql +DATACONVERT_MAX_ROWS= +DATACONVERT_SQL_FORMAT=batch +DATACONVERT_BATCH_ROWS=500 +DATACONVERT_MAX_FILE_SIZE=100M +DATACONVERT_ALL_TEXT=false +DATACONVERT_KEEP_FOLDERS=true +DATACONVERT_NO_SCHEMA=false +DATACONVERT_NO_DDL=false diff --git a/soleprint/station/tools/dataconvert/README.md b/soleprint/station/tools/dataconvert/README.md index 41cf30c..953d788 100644 --- a/soleprint/station/tools/dataconvert/README.md +++ b/soleprint/station/tools/dataconvert/README.md @@ -100,7 +100,7 @@ Shapes are judged on up to 200,000 values spread through the column. ## Folders and exclusions Output is flat by default: a table's file is named after the file or sheet, and two -source folders with the same table name (two studies' `drm_lb.csv`) write into the +source folders with the same table name (two exports' `users.csv`) write into the same file. `keep_folders` mirrors the source folders under `out_dir` instead, so each keeps its own `.sql`; `SCHEMA.md` stays one document at the top, with the folder in each table's heading. diff --git a/soleprint/station/tools/dataconvert/config.py b/soleprint/station/tools/dataconvert/config.py index 536a0cf..2ee99f1 100644 --- a/soleprint/station/tools/dataconvert/config.py +++ b/soleprint/station/tools/dataconvert/config.py @@ -17,6 +17,8 @@ header on row 1. "batch_rows": 500, "max_file_size": "100M", "schema": true, + "include_ddl": true, + "load_order": ["ref_data", "users", "orders"], "header_row": 1, "layouts": [ { @@ -35,11 +37,11 @@ is one of match.in. The first layout that matches wins; a sheet none matches uses the top-level header_row/data_row, which default to 1 and 2. The run settings (out_dir, max_rows, schema, sql_format, batch_rows, -max_file_size, all_text, keep_folders, exclude) are the command line's flags -with the same names, and a flag given on the command line wins over the file; ---exclude patterns are added to the file's. out_dir is relative to the folder -the config file is in; max_file_size takes a number of bytes or a suffixed size -such as "100M" (decimal: k, M, G). Keys starting with _ are +max_file_size, all_text, keep_folders, exclude, include_ddl, load_order) are the +command line's flags with the same names, and a flag given on the command line +wins over the file; --exclude patterns are added to the file's. out_dir is relative +to the folder the config file is in; max_file_size takes a number of bytes or a +suffixed size such as "100M" (decimal: k, M, G). Keys starting with _ are comments; any other unknown key is an error, so a typo is not silently ignored. """ @@ -96,7 +98,8 @@ class Config: def __init__(self, rules=(), bare_sheet_prefixes=(), source=None, forced=None, fallback=None, max_rows=None, schema=None, out_dir=None, sql_format=None, batch_rows=None, max_file_size=None, - all_text=None, keep_folders=None, exclude=()): + all_text=None, keep_folders=None, exclude=(), include_ddl=None, + load_order=()): self.rules = list(rules) self.bare_sheet_prefixes = tuple(bare_sheet_prefixes) self.source = source @@ -115,6 +118,8 @@ class Config: self.all_text = all_text self.keep_folders = keep_folders self.exclude = tuple(exclude) + self.include_ddl = include_ddl + self.load_order = tuple(load_order) @property def probe_rows(self): @@ -137,6 +142,8 @@ class Config: said.append(f"max_rows {self.max_rows}") if self.schema is not None: said.append(f"schema {'on' if self.schema else 'off'}") + if self.include_ddl is not None: + said.append(f"include_ddl {'on' if self.include_ddl else 'off'}") if self.sql_format is not None: said.append(f"sql_format {self.sql_format}") if self.batch_rows is not None: @@ -149,6 +156,8 @@ class Config: said.append("keep_folders") if self.exclude: said.append(f"exclude {', '.join(self.exclude)}") + if self.load_order: + said.append(f"load_order {', '.join(self.load_order)}") if self.fallback is not None: said.append(f"header row {self.fallback.header_row}, data from row {self.fallback.data_row}") if self.rules: @@ -158,7 +167,8 @@ class Config: FILENAME = "dataconvert.json" KNOWN_KEYS = {"layouts", "bare_sheet_prefixes", "max_rows", "schema", "out_dir", "header_row", "data_row", - "sql_format", "batch_rows", "max_file_size", "all_text", "keep_folders", "exclude"} + "sql_format", "batch_rows", "max_file_size", "all_text", "keep_folders", "exclude", + "include_ddl", "load_order"} def parse_size(value): @@ -214,6 +224,9 @@ def load(path=None, forced=None): schema = raw.get("schema") if schema is not None and not isinstance(schema, bool): raise ConfigError(f"{path}: schema must be true or false") + include_ddl = raw.get("include_ddl") + if include_ddl is not None and not isinstance(include_ddl, bool): + raise ConfigError(f"{path}: include_ddl must be true or false") out_dir = raw.get("out_dir") if out_dir is not None: if not isinstance(out_dir, str) or not out_dir.strip(): @@ -242,6 +255,10 @@ def load(path=None, forced=None): if not isinstance(exclude, list) or not all(isinstance(p, str) and p for p in exclude): raise ConfigError(f'{path}: exclude must be a list of patterns such as "*.xlsx.ods"') + load_order = raw.get("load_order", []) + if not isinstance(load_order, list) or not all(isinstance(p, str) and p for p in load_order): + raise ConfigError(f'{path}: load_order must be a list of table identifiers') + fallback = None if "header_row" in raw or "data_row" in raw: try: @@ -252,4 +269,5 @@ def load(path=None, forced=None): return Config(rules, [str(p) for p in prefixes], source=path, forced=forced, fallback=fallback, max_rows=max_rows, schema=schema, out_dir=out_dir, sql_format=sql_format, batch_rows=batch_rows, max_file_size=max_file_size, - all_text=flags["all_text"], keep_folders=flags["keep_folders"], exclude=exclude) + all_text=flags["all_text"], keep_folders=flags["keep_folders"], exclude=exclude, + include_ddl=include_ddl, load_order=load_order) diff --git a/soleprint/station/tools/dataconvert/dataconvert-example.json b/soleprint/station/tools/dataconvert/dataconvert-example.json index 8fdfcfd..775821b 100644 --- a/soleprint/station/tools/dataconvert/dataconvert-example.json +++ b/soleprint/station/tools/dataconvert/dataconvert-example.json @@ -2,7 +2,7 @@ "_comment": "Template for one set of spreadsheets. Copy it to dataconvert.json in the folder that holds them, where it is picked up automatically, or pass any file with --config FILE. Every key is optional; keys starting with _ are comments, any other unknown key is an error. Rows and columns count from 1, as the spreadsheet shows them.", "_run": "The same settings as the command-line flags, and a flag given there wins. out_dir is relative to this file's folder. max_rows samples each table (leave it out for all rows). schema false skips SCHEMA.md. sql_format: insert (a statement per row, default), batch (one INSERT per batch_rows rows), or copy (COPY FROM stdin: smallest, psql only, load once). max_file_size splits a table into numbered parts past that size, e.g. \"100M\".", - "out_dir": "seed", + "out_dir": "datadir/sql", "_values": "all_text reads every cell as the string in the file (only empty cells become NULL), so codes keep leading zeros and NA stays NA; SCHEMA.md then says what each column looks like. keep_folders mirrors the source folders under out_dir, so same-named tables from different folders do not share a file. exclude skips files by glob: no / matches the file name at any depth, with a / the path under the input folder.", "all_text": true, diff --git a/soleprint/station/tools/dataconvert/dataconvert.py b/soleprint/station/tools/dataconvert/dataconvert.py index c58dc82..06bf161 100644 --- a/soleprint/station/tools/dataconvert/dataconvert.py +++ b/soleprint/station/tools/dataconvert/dataconvert.py @@ -25,6 +25,7 @@ it is copied: run this script from anywhere, no install step. """ import argparse +import os from pathlib import Path from output import write_tables @@ -35,6 +36,24 @@ from readers import expand_inputs, iter_sources from schema import SchemaReport +def load_dotenv(): + env_path = Path(".env") + if env_path.is_file(): + for line in env_path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if line and not line.startswith("#") and "=" in line: + key, val = line.split("=", 1) + if key not in os.environ: + os.environ[key] = val.strip().strip('"\'') + + +def get_bool_env(key, default=False): + val = os.getenv(key) + if val is None: + return default + return val.strip().lower() in ('true', '1', 'yes', 'on') + + def size(value): try: return cfg.parse_size(value) @@ -50,35 +69,41 @@ def positive_int(value): def main(): + load_dotenv() + + default_input = os.getenv("DATACONVERT_INPUT") + default_exclude = os.getenv("DATACONVERT_EXCLUDE", "").split(",") if os.getenv("DATACONVERT_EXCLUDE") else [] + parser = argparse.ArgumentParser(description="Convert data sources into schema-agnostic SQL seed files.") - parser.add_argument("--input", nargs="+", required=True, help="Input file(s), directory, wildcard pattern(s), or ZIP archive(s)") - parser.add_argument("--out-dir", default=None, - help="Directory where individual .sql files will be written (default: the config's out_dir; one of the two is required)") - parser.add_argument("--max-rows", type=positive_int, default=None, - help="Write at most N rows per table; the header and SCHEMA.md note the full row count and size (default: the config's max_rows, else all)") - parser.add_argument("--no-schema", action="store_true", help="Do not write SCHEMA.md, whatever the config says") - parser.add_argument("--sql-format", choices=FORMATS, default=None, + parser.add_argument("--input", nargs="+", default=[default_input] if default_input else None, required=not bool(default_input), + help="Input file(s), directory, wildcard pattern(s), or ZIP archive(s)") + parser.add_argument("--out-dir", default=os.getenv("DATACONVERT_OUT_DIR"), + help="Directory where individual .sql files will be written (default: the config's out_dir or env)") + parser.add_argument("--max-rows", type=positive_int, default=os.getenv("DATACONVERT_MAX_ROWS"), + help="Write at most N rows per table; the header and SCHEMA.md note the full row count and size") + parser.add_argument("--no-schema", action="store_true", default=get_bool_env("DATACONVERT_NO_SCHEMA"), + help="Do not write SCHEMA.md, whatever the config says") + parser.add_argument("--no-ddl", action="store_true", default=get_bool_env("DATACONVERT_NO_DDL"), + help="Do not emit CREATE TABLE IF NOT EXISTS statements before seeding") + parser.add_argument("--sql-format", choices=FORMATS, default=os.getenv("DATACONVERT_SQL_FORMAT"), help="insert: a statement per row (default). batch: one INSERT per --batch-rows rows. " "copy: COPY FROM stdin, smallest, psql only, load once") - parser.add_argument("--batch-rows", type=positive_int, default=None, + parser.add_argument("--batch-rows", type=positive_int, default=os.getenv("DATACONVERT_BATCH_ROWS", DEFAULT_BATCH_ROWS), help=f"Rows per INSERT with --sql-format batch (default {DEFAULT_BATCH_ROWS})") - parser.add_argument("--max-file-size", type=size, default=None, - help='Split a table into numbered parts past this size, e.g. "100M" (default: never split; samples are never split)') - parser.add_argument("--all-text", action="store_true", default=None, - help="Read every column as text, exactly as in the file; only empty cells become NULL. " - "Types are then the loader's business; SCHEMA.md says what each column looks like") - parser.add_argument("--keep-folders", action="store_true", default=None, - help="Mirror the source folders under the output directory, instead of one flat folder " - "where same-named tables from different folders share a file") - parser.add_argument("--exclude", action="append", default=[], metavar="GLOB", - help='Skip matching files, e.g. "*.xlsx.ods"; a pattern with no / matches the file name ' - "at any depth. Repeatable, and added to the config's") + parser.add_argument("--max-file-size", type=size, default=os.getenv("DATACONVERT_MAX_FILE_SIZE"), + help='Split a table into numbered parts past this size, e.g. "100M"') + parser.add_argument("--all-text", action="store_true", default=get_bool_env("DATACONVERT_ALL_TEXT", None), + help="Read every column as text, exactly as in the file; only empty cells become NULL.") + parser.add_argument("--keep-folders", action="store_true", default=get_bool_env("DATACONVERT_KEEP_FOLDERS", None), + help="Mirror the source folders under the output directory") + parser.add_argument("--exclude", action="append", default=default_exclude, metavar="GLOB", + help='Skip matching files, e.g. "*.xlsx.ods". Repeatable, and added to the config\'s') parser.add_argument("--header-row", type=positive_int, default=1, help="Spreadsheet row holding the column names, for every file; overrides the config's layouts (default 1)") parser.add_argument("--data-row", type=positive_int, default=None, help="First row of data, when rows sit between it and the header (default: the row after the header)") - parser.add_argument("--config", default=None, - help="JSON with layouts, naming and run settings, for every input (default: dataconvert.json in each input's folder, if there is one)") + parser.add_argument("--config", default=os.getenv("DATACONVERT_CONFIG"), + help="JSON with layouts, naming and run settings, for every input") args = parser.parse_args() forced = None @@ -129,14 +154,16 @@ def main(): # The command line wins, then the config. out_dir = Path(args.out_dir) if args.out_dir else config.out_dir - max_rows = args.max_rows if args.max_rows is not None else config.max_rows + max_rows = int(args.max_rows) if args.max_rows is not None else config.max_rows schema = not args.no_schema and config.schema is not False + include_ddl = not args.no_ddl and config.include_ddl is not False fmt = args.sql_format or config.sql_format or "insert" - batch_rows = args.batch_rows or config.batch_rows or DEFAULT_BATCH_ROWS + batch_rows = int(args.batch_rows) if args.batch_rows else (config.batch_rows or DEFAULT_BATCH_ROWS) max_bytes = args.max_file_size or config.max_file_size all_text = bool(args.all_text or config.all_text) keep_folders = bool(args.keep_folders or config.keep_folders) exclude = tuple(config.exclude) + tuple(args.exclude) + load_order = config.load_order key = out_dir.resolve() if key not in reports: @@ -146,7 +173,8 @@ def main(): for source_name, folder, dfs in iter_sources(in_path, config, all_text, exclude): write_tables(dfs, out_dir, source_name, max_rows, report if schema else None, config.bare_sheet_prefixes, - fmt, batch_rows, max_bytes, folder if keep_folders else "") + fmt, batch_rows, max_bytes, folder if keep_folders else "", include_ddl=include_ddl, + load_order=load_order) for out_dir, report, caps in reports.values(): if report.entries and out_dir.is_dir(): diff --git a/soleprint/station/tools/dataconvert/output.py b/soleprint/station/tools/dataconvert/output.py index efbec72..a5bbe60 100644 --- a/soleprint/station/tools/dataconvert/output.py +++ b/soleprint/station/tools/dataconvert/output.py @@ -11,16 +11,16 @@ from progress import say from sqlgen import DEFAULT_BATCH_ROWS, Shape, render_sample, sanitize_identifier, utf8_len -def table_filename(raw_name: str, source_name: str, sheet_count: int, bare_prefixes=()) -> str: +def table_filename(raw_name: str, source_name: str, sheet_count: int, bare_prefixes=(), prefix="") -> str: """ Sheets of a multi-sheet workbook are prefixed with the workbook, so two workbooks cannot collide, unless the config names the sheet as already - unique (bare_sheet_prefixes). + unique (bare_sheet_prefixes). An optional numeric prefix (e.g. 01_) forces + strict load order. """ clean = sanitize_identifier(raw_name) - if sheet_count > 1 and not clean.startswith(tuple(bare_prefixes)): - return f"{sanitize_identifier(source_name)}_{clean}.sql" - return f"{clean}.sql" + base_name = f"{sanitize_identifier(source_name)}_{clean}.sql" if sheet_count > 1 and not clean.startswith(tuple(bare_prefixes)) else f"{clean}.sql" + return f"{prefix}{base_name}" if prefix else base_name class TableFile: @@ -111,19 +111,34 @@ def write_full(shape: Shape, df, table_file: TableFile) -> int: def write_tables(dfs: dict, out_dir: Path, source_name: str, max_rows=None, report=None, bare_prefixes=(), - fmt="insert", batch_rows=DEFAULT_BATCH_ROWS, max_bytes=None, folder=""): + fmt="insert", batch_rows=DEFAULT_BATCH_ROWS, max_bytes=None, folder="", include_ddl=True, + load_order=()): """ Write individual .sql files per table/sheet into the output directory, or into its `folder` subdirectory when the source folders are kept. + Respects DDL-first generation and load_order configuration. """ target = out_dir / folder if folder else out_dir - for raw_name, df in dfs.items(): + + order_map = {sanitize_identifier(name): idx for idx, name in enumerate(load_order, start=1)} + + def table_sort_key(item): + raw_name = item[0] + sanitized = sanitize_identifier(raw_name) + return (order_map.get(sanitized, 9999), sanitized) + + sorted_dfs = sorted(dfs.items(), key=table_sort_key) + + for raw_name, df in sorted_dfs: if df.empty: continue table = sanitize_identifier(raw_name) - filename = table_filename(raw_name, source_name, len(dfs), bare_prefixes) - shape = Shape(fmt, table, df.columns, batch_rows) + order_idx = order_map.get(table) + file_prefix = f"{order_idx:02d}_" if order_idx is not None else "" + + filename = table_filename(raw_name, source_name, len(dfs), bare_prefixes, prefix=file_prefix) + shape = Shape(fmt, table, df.columns, batch_rows, include_ddl=include_ddl, df=df) total = len(df) exact = max_rows is None or max_rows >= total # A sample is never split: it is small, and it is not for loading. diff --git a/soleprint/station/tools/dataconvert/sqlgen.py b/soleprint/station/tools/dataconvert/sqlgen.py index 8b96299..3554d9a 100644 --- a/soleprint/station/tools/dataconvert/sqlgen.py +++ b/soleprint/station/tools/dataconvert/sqlgen.py @@ -38,6 +38,33 @@ def sanitize_identifier(identifier: str) -> str: return clean.strip("_") +def postgres_type(series: pd.Series) -> str: + """Infer a PostgreSQL column data type for DDL generation.""" + vals = series.dropna() + if vals.empty: + return "TEXT" + if pd.api.types.is_bool_dtype(vals): + return "BOOLEAN" + if pd.api.types.is_integer_dtype(vals): + return "BIGINT" + if pd.api.types.is_float_dtype(vals): + return "NUMERIC" + if pd.api.types.is_datetime64_any_dtype(vals): + return "TIMESTAMP" + return "TEXT" + + +def ddl_statement(table_name: str, df: pd.DataFrame) -> str: + """Emit DDL for table creation before seeding data so that loading does not fail.""" + cols_def = [] + for col in df.columns: + col_id = sanitize_identifier(col) + col_type = postgres_type(df[col]) + cols_def.append(f' "{col_id}" {col_type}') + cols_sql = ",\n".join(cols_def) + return f'CREATE TABLE IF NOT EXISTS "{table_name}" (\n{cols_sql}\n);\n\n' + + def sql_value(v) -> str: if pd.isna(v): return "NULL" @@ -68,12 +95,14 @@ def utf8_len(text: str) -> int: class Shape: """How one table's rows are framed in one format.""" - def __init__(self, fmt, table_name, columns, batch_rows=DEFAULT_BATCH_ROWS): + def __init__(self, fmt, table_name, columns, batch_rows=DEFAULT_BATCH_ROWS, include_ddl=True, df=None): if fmt not in FORMATS: raise ValueError(f"unknown sql format: {fmt}") self.fmt = fmt self.table_name = table_name self.batch_rows = batch_rows + self.include_ddl = include_ddl + self.df = df table_ref = f'"{table_name}"' cols = ", ".join(f'"{sanitize_identifier(c)}"' for c in columns) self.head = f"-- Generated seed data for table: {table_ref}\n" @@ -84,6 +113,8 @@ class Shape: def open(self, note="") -> str: text = self.head + note + "BEGIN;\n\n" + if self.include_ddl and self.df is not None: + text += ddl_statement(self.table_name, self.df) return text + self.copy if self.fmt == "copy" else text def close(self) -> str: diff --git a/soleprint/station/tools/dataconvert/uv.lock b/soleprint/station/tools/dataconvert/uv.lock index 1a6ed26..4fc2029 100644 --- a/soleprint/station/tools/dataconvert/uv.lock +++ b/soleprint/station/tools/dataconvert/uv.lock @@ -13,6 +13,23 @@ resolution-markers = [ "python_full_version < '3.12' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] +[[package]] +name = "dataconvert" +version = "0.1.0" +source = { virtual = "." } +dependencies = [ + { name = "odfpy" }, + { name = "openpyxl" }, + { name = "pandas" }, +] + +[package.metadata] +requires-dist = [ + { name = "odfpy", specifier = ">=1.4.1" }, + { name = "openpyxl", specifier = ">=3.1.2" }, + { name = "pandas", specifier = ">=2.2.0" }, +] + [[package]] name = "defusedxml" version = "0.7.1" @@ -284,23 +301,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" }, ] -[[package]] -name = "dataconvert" -version = "0.1.0" -source = { virtual = "." } -dependencies = [ - { name = "odfpy" }, - { name = "openpyxl" }, - { name = "pandas" }, -] - -[package.metadata] -requires-dist = [ - { name = "odfpy", specifier = ">=1.4.1" }, - { name = "openpyxl", specifier = ">=3.1.2" }, - { name = "pandas", specifier = ">=2.2.0" }, -] - [[package]] name = "six" version = "1.17.0" diff --git a/soleprint/station/tools/datasource/.env-example b/soleprint/station/tools/datasource/.env-example new file mode 100644 index 0000000..71176e6 --- /dev/null +++ b/soleprint/station/tools/datasource/.env-example @@ -0,0 +1,20 @@ +# Environment Variables for datasource tools + +# Database Credentials +POSTGRES_USER=app +POSTGRES_PASSWORD=secret +POSTGRES_DB=app + +# Full Database URI (Overrides user/pass/db if provided) +TARGET_DB_URI=postgresql://app:secret@postgres:5432/app + +# Data Dump configuration +DUMP_OUT_DIR=/backups/cleaned_db_dump +DUMP_IN_DIR=/backups/cleaned_db_dump +# Pass schemas exactly as required by pg_dump, e.g., "--schema=core --schema=sales" +DUMP_SCHEMAS="--schema=core" + +# Data Load / Seeding Configuration +SEED_DIR=/rig/datadir/sql +TOOLS_DIR=/rig/tools/datasource +CONF_DIR=/rig/datadir/datasource diff --git a/soleprint/station/tools/datasource/.gitignore b/soleprint/station/tools/datasource/.gitignore new file mode 100644 index 0000000..9e3af16 --- /dev/null +++ b/soleprint/station/tools/datasource/.gitignore @@ -0,0 +1,4 @@ +# The real mapping: how one producer's raw tables become the logical domains names +# their tables and columns. It lives in the data folder (datadir/datasource/), not +# here; this only stops a stray copy being committed. Start from schema_mapping-example.yaml. +schema_mapping.yaml diff --git a/soleprint/station/tools/datasource/README.md b/soleprint/station/tools/datasource/README.md new file mode 100644 index 0000000..c3ca531 --- /dev/null +++ b/soleprint/station/tools/datasource/README.md @@ -0,0 +1,26 @@ +# datasource + +Scripts to dump, restore, and seed the database, including schema migration mapping. + +## Schema Mapping + +The data loading process converges raw seed tables into logical domains using a YAML configuration. +This configuration is injected at runtime from `datadir/datasource/schema_mapping.yaml` (never +committed: it names the producer's tables) rather than hardcoded in the tool. Start from +`schema_mapping-example.yaml`. + +Example mapping structure: + +```yaml +tables: + - source: raw_users + target: core.users + truncate: true + columns: + user_id: id + username: name + created_at: etl_utils.safe_cast_date(created_date) + is_active: "COALESCE(etl_utils.safe_cast_boolean(status), TRUE)" +``` + +All operational variables (database credentials, dump directories, active schemas) are read from the environment or a `.env` file at the root. See `.env-example` for available configuration keys. diff --git a/soleprint/station/tools/datasource/casting_helpers.sql b/soleprint/station/tools/datasource/casting_helpers.sql new file mode 100644 index 0000000..d3c7c1c --- /dev/null +++ b/soleprint/station/tools/datasource/casting_helpers.sql @@ -0,0 +1,30 @@ +-- Create dedicated schema for ETL migration helpers +CREATE SCHEMA IF NOT EXISTS etl_utils; + +-- 1. Safe String to Boolean Converter +CREATE OR REPLACE FUNCTION etl_utils.safe_cast_boolean(val TEXT) +RETURNS BOOLEAN AS $$ BEGIN IF val IS NULL OR TRIM(val) = '' OR UPPER(TRIM(val)) IN ('NULL', 'N/A', '.') THEN RETURN NULL; ELSIF UPPER(TRIM(val)) IN ('Y', 'YES', '1', 'TRUE') THEN RETURN TRUE; ELSIF UPPER(TRIM(val)) IN ('N', 'NO', '0', 'FALSE') THEN RETURN FALSE; ELSE RETURN NULL; END IF; END; $$ LANGUAGE plpgsql IMMUTABLE; + +-- 2. Safe String to Numeric/Integer Converter +CREATE OR REPLACE FUNCTION etl_utils.safe_cast_numeric(val TEXT) +RETURNS NUMERIC AS $$ DECLARE cleaned TEXT; BEGIN cleaned := REGEXP_REPLACE(TRIM(val), '[^0-9.-]', '', 'g'); IF cleaned = '' OR cleaned = '.' OR cleaned = '-' THEN RETURN NULL; END IF; RETURN cleaned::NUMERIC; EXCEPTION WHEN OTHERS THEN RETURN NULL; END; $$ LANGUAGE plpgsql IMMUTABLE; + +-- 3. Safe String to Date/Timestamp Converter +CREATE OR REPLACE FUNCTION etl_utils.safe_cast_date(val TEXT) +RETURNS DATE AS $func$ +BEGIN + IF val IS NULL OR TRIM(val) = '' OR val ~* '^(UN|UNK|UNKNOWN)' THEN + RETURN NULL; + END IF; + + BEGIN + RETURN val::DATE; + EXCEPTION WHEN OTHERS THEN + BEGIN + RETURN TO_DATE(val, 'DD-MON-YY'); + EXCEPTION WHEN OTHERS THEN + RETURN NULL; + END; + END; +END; +$func$ LANGUAGE plpgsql IMMUTABLE; \ No newline at end of file diff --git a/soleprint/station/tools/datasource/dump_trasformed.sh b/soleprint/station/tools/datasource/dump_trasformed.sh new file mode 100644 index 0000000..e56d9bf --- /dev/null +++ b/soleprint/station/tools/datasource/dump_trasformed.sh @@ -0,0 +1,20 @@ +#!/bin/bash +set -euo pipefail + +# Load environment variables from .env if present +if [ -f .env ]; then + export $(grep -v '^#' .env | xargs) +fi + +DB_USER="${POSTGRES_USER:-postgres}" +DB_NAME="${POSTGRES_DB:-production_db}" +OUT_DIR="${DUMP_OUT_DIR:-/backups/cleaned_db_dump}" +SCHEMAS="${DUMP_SCHEMAS:-}" + +# Dump only the transformed, strictly-typed schemas (skipping raw staging tables) +# Note: Pass schemas in DUMP_SCHEMAS as e.g. "--schema=schema1 --schema=schema2" +pg_dump -U "$DB_USER" -d "$DB_NAME" \ + ${SCHEMAS} \ + --format=directory \ + --jobs=4 \ + --file="$OUT_DIR" diff --git a/soleprint/station/tools/datasource/load_seed_data.sh b/soleprint/station/tools/datasource/load_seed_data.sh new file mode 100644 index 0000000..2694565 --- /dev/null +++ b/soleprint/station/tools/datasource/load_seed_data.sh @@ -0,0 +1,59 @@ +#!/bin/bash +set -euo pipefail + +# Load environment variables from .env if present +if [ -f .env ]; then + export $(grep -v '^#' .env | xargs) +fi + +# Build the URI dynamically using the injected Postgres secrets +# Fall back to the default 'app' user if the variables are missing +USER="${POSTGRES_USER:-app}" +PASS="${POSTGRES_PASSWORD:-}" +DB="${POSTGRES_DB:-app}" + +DB_URI="${TARGET_DB_URI:-postgresql://${USER}:${PASS}@postgres:5432/${DB}}" + +# Point to the containerPaths mapped in your kind-config.yaml.tpl +SEED_DIR="${SEED_DIR:-/rig/datadir/sql}" +TOOLS_DIR="${TOOLS_DIR:-/rig/tools/datasource}" +CONF_DIR="${CONF_DIR:-/rig/datadir/datasource}" + +echo "Optimizing PostgreSQL for bulk load..." +psql -v ON_ERROR_STOP=1 "$DB_URI" <<-EOSQL + ALTER SYSTEM SET maintenance_work_mem = '1GB'; + ALTER SYSTEM SET synchronous_commit = 'off'; + ALTER SYSTEM SET checkpoint_timeout = '15min'; + ALTER SYSTEM SET max_wal_size = '2GB'; + SELECT pg_reload_conf(); +EOSQL + +echo "Loading casting helpers..." +psql -v ON_ERROR_STOP=1 "$DB_URI" -f "$TOOLS_DIR/casting_helpers.sql" + +echo "Loading seed data files in single transactions..." +if [ -d "$SEED_DIR" ]; then + for f in "$SEED_DIR"/*.sql; do + if [ -f "$f" ]; then + echo "Processing $f..." + psql -v ON_ERROR_STOP=1 -1 "$DB_URI" -f "$f" + fi + done +else + echo "Warning: Seed directory $SEED_DIR not found. Skipping flat file loads." +fi + +echo "Converging schemas into logical domains using configured mapping..." +MAPPING_FILE="${SCHEMA_MAPPING_FILE:-$CONF_DIR/schema_mapping.yaml}" +python3 "$TOOLS_DIR/migrate_schema.py" "$MAPPING_FILE" + +echo "Restoring standard PostgreSQL settings..." +psql -v ON_ERROR_STOP=1 "$DB_URI" <<-EOSQL + ALTER SYSTEM RESET maintenance_work_mem; + ALTER SYSTEM RESET synchronous_commit; + ALTER SYSTEM RESET checkpoint_timeout; + ALTER SYSTEM RESET max_wal_size; + SELECT pg_reload_conf(); +EOSQL + +echo "Data load completed." diff --git a/soleprint/station/tools/datasource/migrate_schema.py b/soleprint/station/tools/datasource/migrate_schema.py new file mode 100644 index 0000000..c7ce5b9 --- /dev/null +++ b/soleprint/station/tools/datasource/migrate_schema.py @@ -0,0 +1,54 @@ +import os +import sys +import yaml +import psycopg +from psycopg import sql + +def main(): + mapping_file = os.getenv("SCHEMA_MAPPING_FILE", "/rig/datadir/datasource/schema_mapping.yaml") + + if len(sys.argv) >= 2: + mapping_file = sys.argv[1] + + db_uri = os.getenv("TARGET_DB_URI", "postgresql://postgres:postgres@postgres:5432/postgres") + + print(f"Loading schema mapping from {mapping_file}...") + with open(mapping_file, 'r') as f: + config = yaml.safe_load(f) + + with psycopg.connect(db_uri) as conn: + with conn.cursor() as cur: + for table in config.get('tables', []): + source = table['source'] + target = table['target'] + columns = table['columns'] + + print(f"Migrating {source} to {target}...") + + if table.get('truncate', False): + try: + cur.execute(f"TRUNCATE TABLE {target} RESTART IDENTITY CASCADE;") + except psycopg.errors.UndefinedTable: + print(f" Target table {target} does not exist yet; skipping truncate.") + conn.rollback() + + target_cols = ", ".join(columns.keys()) + source_exprs = ", ".join(columns.values()) + + query = f""" + INSERT INTO {target} ({target_cols}) + SELECT {source_exprs} + FROM {source} + """ + try: + cur.execute(query) + print(f" Successfully migrated {source}.") + except Exception as e: + print(f" Failed to migrate {source}: {e}") + conn.rollback() + raise + conn.commit() + print("Schema migration complete.") + +if __name__ == '__main__': + main() diff --git a/soleprint/station/tools/datasource/restore_transformed.sh b/soleprint/station/tools/datasource/restore_transformed.sh new file mode 100644 index 0000000..170077a --- /dev/null +++ b/soleprint/station/tools/datasource/restore_transformed.sh @@ -0,0 +1,18 @@ +#!/bin/bash +set -euo pipefail + +# Load environment variables from .env if present +if [ -f .env ]; then + export $(grep -v '^#' .env | xargs) +fi + +DB_USER="${POSTGRES_USER:-postgres}" +DB_NAME="${POSTGRES_DB:-production_db}" +IN_DIR="${DUMP_IN_DIR:-/backups/cleaned_db_dump}" + +# Multi-threaded restore into the target PostgreSQL cluster pod +pg_restore -U "$DB_USER" -d "$DB_NAME" \ + --jobs=4 \ + --clean \ + --if-exists \ + "$IN_DIR" diff --git a/soleprint/station/tools/datasource/schema_mapping-example.yaml b/soleprint/station/tools/datasource/schema_mapping-example.yaml new file mode 100644 index 0000000..85ec61b --- /dev/null +++ b/soleprint/station/tools/datasource/schema_mapping-example.yaml @@ -0,0 +1,21 @@ +# Schema Conversion and Type Cast Registry +# Maps raw seed tables to the unified logical domains. + +tables: + - source: raw_users + target: core.users + truncate: true + columns: + user_id: id + username: name + created_at: etl_utils.safe_cast_date(created_date) + is_active: "COALESCE(etl_utils.safe_cast_boolean(status), TRUE)" + + - source: raw_orders + target: core.orders + truncate: true + columns: + order_id: id + user_id: user_id + total_amount: etl_utils.safe_cast_numeric(amount) + order_date: etl_utils.safe_cast_date(date)