distilled tools

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

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.

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)

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,

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():

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.

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:

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"

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

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

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.

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;

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"

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."

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()

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"

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)