From 55c0d73ffe07fe60e1cf18ea710e9a34f765ed27 Mon Sep 17 00:00:00 2001 From: buenosairesam Date: Tue, 22 Sep 2026 11:57:31 -0300 Subject: [PATCH] update dataconvert and source --- .../station/tools/dataconvert/readers.py | 26 ++++++++++++++++--- soleprint/station/tools/dataconvert/sqlgen.py | 8 +++++- soleprint/station/tools/datasource/README.md | 4 +++ .../tools/datasource/migrate_schema.py | 18 ++++++++----- 4 files changed, 45 insertions(+), 11 deletions(-) diff --git a/soleprint/station/tools/dataconvert/readers.py b/soleprint/station/tools/dataconvert/readers.py index 0a593a4..33b3e86 100644 --- a/soleprint/station/tools/dataconvert/readers.py +++ b/soleprint/station/tools/dataconvert/readers.py @@ -17,7 +17,7 @@ from pathlib import Path, PurePosixPath import pandas as pd from progress import say, seconds -from sqlgen import human_bytes +from sqlgen import human_bytes, sanitize_identifier SUPPORTED = {".csv", ".xlsx", ".xls", ".ods"} @@ -68,13 +68,33 @@ def read_sheet(read, config): if layout is None: layout = config.fallback if layout is None: - return read(), None + data = read() + data.columns = usable_columns(data.columns) + return data, None raw = read(header=None) data = raw.iloc[layout.data_row - 1:].copy() - data.columns = [str(c).strip() for c in raw.iloc[layout.header_row - 1]] + data.columns = usable_columns(str(c).strip() for c in raw.iloc[layout.header_row - 1]) return data, layout +def usable_columns(names) -> list: + """ + Header cells as SQL can take them: an empty one (or one of symbols only, which + sanitizes to nothing) becomes column_, n counted from 1 as the spreadsheet does, + and a name that repeats once sanitized gets _2, _3. Every other name is kept as written. + """ + out, seen = [], set() + for n, name in enumerate(names, start=1): + name = str(name) + base = name if sanitize_identifier(name) else f"column_{n}" + candidate, k = base, 2 + while sanitize_identifier(candidate) in seen: + candidate, k = f"{base}_{k}", k + 1 + seen.add(sanitize_identifier(candidate)) + out.append(candidate) + return out + + def load_dataframes_from_file(file_path: Path, config, all_text=False) -> dict: """Load a file (.csv, .xlsx, .xls, .ods) into {entity_name: DataFrame}.""" ext = file_path.suffix.lower() diff --git a/soleprint/station/tools/dataconvert/sqlgen.py b/soleprint/station/tools/dataconvert/sqlgen.py index 3554d9a..a4b9783 100644 --- a/soleprint/station/tools/dataconvert/sqlgen.py +++ b/soleprint/station/tools/dataconvert/sqlgen.py @@ -62,7 +62,11 @@ def ddl_statement(table_name: str, df: pd.DataFrame) -> str: 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' + # Several sources can feed one table (same-named files in different folders), each + # block with its own CREATE; only the first creates. Each block then adds the columns + # only it has, so the table ends up with every source's columns, NULL where one lacks them. + adds = "".join(f'ALTER TABLE "{table_name}" ADD COLUMN IF NOT EXISTS{line};\n' for line in cols_def) + return f'CREATE TABLE IF NOT EXISTS "{table_name}" (\n{cols_sql}\n);\n{adds}\n' def sql_value(v) -> str: @@ -114,6 +118,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: + # IF NOT EXISTS says "skipping" once per column per block; keep the load log to errors. + text += "SET LOCAL client_min_messages = warning;\n" text += ddl_statement(self.table_name, self.df) return text + self.copy if self.fmt == "copy" else text diff --git a/soleprint/station/tools/datasource/README.md b/soleprint/station/tools/datasource/README.md index c3ca531..a2dcc01 100644 --- a/soleprint/station/tools/datasource/README.md +++ b/soleprint/station/tools/datasource/README.md @@ -23,4 +23,8 @@ tables: is_active: "COALESCE(etl_utils.safe_cast_boolean(status), TRUE)" ``` +A target that does not exist yet is created from the mapping: its schema, then a table whose +columns are the mapping's names, typed by its expressions (`safe_cast_date` gives a `date`). A +target that already exists, from a real DDL with keys and constraints, is used as it is. + 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/migrate_schema.py b/soleprint/station/tools/datasource/migrate_schema.py index c7ce5b9..5ab7ee7 100644 --- a/soleprint/station/tools/datasource/migrate_schema.py +++ b/soleprint/station/tools/datasource/migrate_schema.py @@ -25,16 +25,20 @@ def main(): 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()) + # A missing target is made from the mapping itself: its schema, then a table + # shaped by the expressions (safe_cast_date gives a date column, and so on). + # A target that exists, from a real DDL with keys and constraints, is left alone. + if "." in target: + cur.execute(f"CREATE SCHEMA IF NOT EXISTS {target.split('.', 1)[0]};") + shaped = ", ".join(f"{expr} AS {col}" for col, expr in columns.items()) + cur.execute(f"CREATE TABLE IF NOT EXISTS {target} AS SELECT {shaped} FROM {source} WITH NO DATA;") + + if table.get('truncate', False): + cur.execute(f"TRUNCATE TABLE {target} RESTART IDENTITY CASCADE;") + query = f""" INSERT INTO {target} ({target_cols}) SELECT {source_exprs}