dashingest

35@dataclass
36class ADLSSource:
37    """Azure Data Lake Storage Gen2 — abfss://<container>@<account>.dfs.core.windows.net/<path>."""
38    storage_account: str
39    container: str
40    path: str = ""
41    file_format: str | None = None
42    reader_options: Any = None
43    options: dict = field(default_factory=dict)

82@dataclass
83class AvroReaderOptions:
84    pass

32@dataclass
33class ConnectionTestResult:
34    ok: bool
35    message: str
36
37    def display(self) -> None:
38        print(f"{'OK' if self.ok else 'Error'} — {self.message}")

28@dataclass
29class CsvReaderOptions:
30    header: bool = True
31    infer_schema: bool = True
32    delimiter: str = ","
33    quote_char: str = '"'
34    escape_char: str = "\\"
35    encoding: str = "UTF-8"
36    null_value: str = ""
37    date_format: str = ""
38    timestamp_format: str = ""
39    multiline: bool = False
40    comment_char: str = ""
41    # PERMISSIVE keeps malformed rows (nulled out) instead of dropping/failing.
42    parse_mode: str = "PERMISSIVE"  # PERMISSIVE | DROPMALFORMED | FAILFAST

56@dataclass
57class DBFSSource:
58    """Legacy DBFS mount or path — dbfs:/<path>."""
59    path: str
60    file_format: str | None = None
61    reader_options: Any = None
62    options: dict = field(default_factory=dict)

 74@dataclass
 75class DatabaseSource:
 76    """A relational database table or query.
 77
 78    Set `engine` + `host` + `database` for a known engine (builds the JDBC
 79    URL/driver for you), or set `url`/`driver` directly for anything else.
 80    Set exactly one of `table` or `query`.
 81
 82    Advanced (all optional): `fetch_size` batches network round-trips;
 83    `num_partitions` + `partition_column` + `lower_bound` + `upper_bound`
 84    split a large table into parallel reads (all four must be set together);
 85    `ssl` sets the common per-engine SSL flag; `connection_properties` is a
 86    raw escape hatch for anything else the JDBC driver accepts.
 87    """
 88    host: str = ""
 89    database: str = ""
 90    engine: str = "postgresql"  # postgresql | mysql | sqlserver | oracle | snowflake
 91    port: int | None = None
 92    table: str = ""
 93    query: str = ""
 94    user: str = ""
 95    password: str = ""
 96    url: str = ""       # overrides the engine preset if set
 97    driver: str = ""    # overrides the engine preset if set
 98    fetch_size: int | None = None
 99    num_partitions: int | None = None
100    partition_column: str = ""
101    lower_bound: int | None = None
102    upper_bound: int | None = None
103    ssl: bool = False
104    connection_properties: dict = field(default_factory=dict)
105
106    @property
107    def is_partitioned(self) -> bool:
108        return bool(self.num_partitions and self.partition_column and self.lower_bound is not None and self.upper_bound is not None)

57@dataclass
58class ExcelReaderOptions:
59    """Comprehensive spark-excel option coverage — the format that actually
60    needs it, since a raw file path alone doesn't tell Spark which sheet,
61    where the header row is, or whether the workbook is password-protected."""
62    sheet_name: str = "0"          # sheet name, or 0-based index as a string
63    header: bool = True
64    header_row: int = 0            # 0-indexed row the header lives on (title rows above it are skipped)
65    data_address: str = ""         # explicit cell range, e.g. "'Sheet1'!B2:F100" — overrides sheet_name/header_row
66    infer_schema: bool = True
67    treat_empty_as_null: bool = True
68    date_format: str = ""
69    timestamp_format: str = ""
70    max_rows_in_memory: int | None = None  # set for large .xlsx files to stream instead of loading fully
71    workbook_password: str = ""
72    # Read and union multiple named sheets with matching schemas into one
73    # DataFrame, instead of a single sheet_name/data_address.
74    sheet_names: list[str] = field(default_factory=list)

22@dataclass
23class IngestResult:
24    table: str
25    row_count: int
26    write_mode: str
27
28    def display(self) -> None:
29        print(f"Ingested into {self.table} ({self.write_mode}): {self.row_count:,} total rows")

142@dataclass
143class IngestTarget:
144    table: str
145    write_mode: str = "append"  # append | overwrite | merge
146    schema_evolution: bool = True
147    merge_keys: list[str] = field(default_factory=list)

45@dataclass
46class JsonReaderOptions:
47    # False = one JSON object per line (JSON Lines); True = pretty-printed /
48    # multi-line records, which Spark can't split on newlines.
49    multiline: bool = False
50    infer_schema: bool = True
51    date_format: str = ""
52    timestamp_format: str = ""
53    parse_mode: str = "PERMISSIVE"
54    primitives_as_string: bool = False

87@dataclass
88class OrcReaderOptions:
89    merge_schema: bool = False

77@dataclass
78class ParquetReaderOptions:
79    merge_schema: bool = False

111@dataclass
112class RestApiSource:
113    """A JSON REST API. `json_path` is a dot-path to the records array/dict
114    if the payload wraps it (e.g. "data.items"); leave empty for a bare array.
115
116    Advanced (all optional): `auth_type` in none|bearer|api_key|basic sets
117    how credentials are attached. `pagination` in none|page_param|cursor
118    follows multiple pages automatically, up to `max_pages`.
119    """
120    url: str
121    headers: dict = field(default_factory=dict)
122    params: dict = field(default_factory=dict)
123    json_path: str = ""
124    timeout_seconds: int = 30
125
126    auth_type: str = "none"  # none | bearer | api_key | basic
127    bearer_token: str = ""
128    api_key_header: str = "X-API-Key"
129    api_key: str = ""
130    basic_user: str = ""
131    basic_password: str = ""
132
133    pagination: str = "none"  # none | page_param | cursor
134    page_param: str = "page"
135    page_size_param: str = ""
136    page_size: int = 0
137    max_pages: int = 20
138    cursor_param: str = "cursor"
139    cursor_json_path: str = ""  # dot-path in the response to the next page's cursor value

46@dataclass
47class S3Source:
48    """Amazon S3 — s3://<bucket>/<path>."""
49    bucket: str
50    path: str = ""
51    file_format: str | None = None
52    reader_options: Any = None
53    options: dict = field(default_factory=dict)

92@dataclass
93class TextReaderOptions:
94    line_sep: str = ""
95    whole_text: bool = False
96    encoding: str = "UTF-8"

23@dataclass
24class VolumeSource:
25    """A Unity Catalog Volume — /Volumes/<catalog>/<schema>/<volume>/<path>."""
26    catalog: str
27    schema_name: str
28    volume: str
29    path: str = ""
30    file_format: str | None = None  # inferred from path extension if omitted
31    reader_options: Any = None      # a CsvReaderOptions/ExcelReaderOptions/... — defaults used if unset
32    options: dict = field(default_factory=dict)  # raw Spark options, applied on top of reader_options

190def build_jdbc_url(source: DatabaseSource) -> str:
191    if source.url:
192        return source.url
193    if source.engine not in JDBC_PRESETS:
194        raise ValueError(f"Unknown engine {source.engine!r}; set url/driver directly for custom engines.")
195    preset = JDBC_PRESETS[source.engine]
196    port = source.port or preset["default_port"]
197    return preset["url"].format(host=source.host, port=port, database=source.database)

 9def env_setup() -> None:
10    """Open the environment setup panel — where should dashingest read/write
11    its configs? Defaults to the notebook's current working directory if
12    never called."""
13    try:
14        import dashui
15        from IPython.display import display
16    except ImportError:
17        raise RuntimeError("ipywidgets required. Run: %pip install ipywidgets") from None
18
19    display(dashui.card([
20        dashui.header("DashIngest — Environment Setup", library=_LIBRARY),
21        dashui.env_setup_panel(_LIBRARY).widget,
22    ]))

153def infer_format_from_path(path: str) -> str | None:
154    ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
155    return FILE_EXTENSIONS.get(ext)

 25def launch():
 26    try:
 27        import ipywidgets as w
 28        from IPython.display import display
 29    except ImportError:
 30        raise RuntimeError("ipywidgets required. Run: %pip install ipywidgets") from None
 31
 32    import dashui
 33
 34    kind_toggle = w.ToggleButtons(
 35        options=["Databricks Volume", "ADLS Gen2", "Amazon S3", "DBFS", "Database", "REST API"],
 36        description="Source:",
 37    )
 38
 39    # ── Databricks Volume ───────────────────────────────────────────────────
 40    vol_catalog = w.Text(description="Catalog:")
 41    vol_schema = w.Text(description="Schema:")
 42    vol_volume = w.Text(description="Volume:")
 43    vol_path = w.Text(description="Path:", placeholder="folder/file.csv (optional)")
 44    vol_box = w.VBox([w.HBox([vol_catalog, vol_schema, vol_volume]), vol_path])
 45
 46    # ── ADLS Gen2 ────────────────────────────────────────────────────────────
 47    adls_account = w.Text(description="Storage account:")
 48    adls_container = w.Text(description="Container:")
 49    adls_path = w.Text(description="Path:", placeholder="folder/file.csv (optional)")
 50    adls_box = w.VBox([w.HBox([adls_account, adls_container]), adls_path])
 51
 52    # ── Amazon S3 ────────────────────────────────────────────────────────────
 53    s3_bucket = w.Text(description="Bucket:")
 54    s3_path = w.Text(description="Path:", placeholder="folder/file.csv (optional)")
 55    s3_box = w.VBox([s3_bucket, s3_path])
 56
 57    # ── DBFS ─────────────────────────────────────────────────────────────────
 58    dbfs_path = w.Text(description="Path:", placeholder="folder/file.csv")
 59    dbfs_box = w.VBox([dbfs_path])
 60
 61    # ── Database ─────────────────────────────────────────────────────────────
 62    db_engine = w.Dropdown(options=["postgresql", "mysql", "sqlserver", "oracle", "snowflake"], description="Engine:")
 63    db_host = w.Text(description="Host:")
 64    db_database = w.Text(description="Database:")
 65    db_port = w.Text(description="Port:", placeholder="default for engine")
 66    db_table = w.Text(description="Table:", placeholder="schema.table")
 67    db_query = w.Text(description="or Query:", placeholder="SELECT ... (instead of table)")
 68    db_user = w.Text(description="User:")
 69    db_password = w.Password(description="Password:")
 70
 71    db_ssl = w.Checkbox(value=False, description="Use SSL")
 72    db_fetch_size = w.IntText(description="Fetch size:", value=0, layout=w.Layout(width="180px"))
 73    db_num_partitions = w.IntText(description="Partitions:", value=0, layout=w.Layout(width="180px"))
 74    db_partition_col = w.Text(description="Partition col:", placeholder="id")
 75    db_lower_bound = w.IntText(description="Lower bound:", layout=w.Layout(width="180px"))
 76    db_upper_bound = w.IntText(description="Upper bound:", layout=w.Layout(width="180px"))
 77    db_props_table = dashui.editable_table(["Property", "Value"], placeholders={"Property": "sslmode", "Value": "require"})
 78    db_advanced = w.Accordion(children=[w.VBox([
 79        db_ssl,
 80        w.HTML("<div style='font-size:12px;color:#5A6872;margin:6px 0 2px'>Parallel read (set all four to split a large table across partitions)</div>"),
 81        w.HBox([db_num_partitions, db_partition_col]),
 82        w.HBox([db_lower_bound, db_upper_bound]),
 83        db_fetch_size,
 84        w.HTML("<div style='font-size:12px;color:#5A6872;margin:6px 0 2px'>Extra JDBC connection properties</div>"),
 85        db_props_table.widget,
 86    ])])
 87    db_advanced.set_title(0, "Advanced")
 88    db_advanced.selected_index = None
 89
 90    db_box = w.VBox([
 91        w.HBox([db_engine, db_host, db_port]),
 92        db_database, db_table, db_query,
 93        w.HBox([db_user, db_password]),
 94        db_advanced,
 95    ])
 96
 97    # ── REST API ─────────────────────────────────────────────────────────────
 98    api_url = w.Text(description="URL:", placeholder="https://api.example.com/records")
 99    api_json_path = w.Text(description="JSON path:", placeholder="data.items (optional)")
100
101    api_auth_type = w.Dropdown(options=["none", "bearer", "api_key", "basic"], description="Auth:")
102    api_bearer_token = w.Password(description="Token:", disabled=True)
103    api_key_header = w.Text(description="Header name:", value="X-API-Key", disabled=True)
104    api_key_value = w.Password(description="Key:", disabled=True)
105    api_basic_user = w.Text(description="User:", disabled=True)
106    api_basic_password = w.Password(description="Password:", disabled=True)
107    api_auth_box = w.VBox([api_bearer_token])
108
109    def on_auth_type_change(change):
110        api_auth_box.children = {
111            "bearer": [api_bearer_token],
112            "api_key": [api_key_header, api_key_value],
113            "basic": [api_basic_user, api_basic_password],
114        }.get(change["new"], [])
115
116    api_auth_type.observe(on_auth_type_change, names="value")
117
118    api_pagination = w.Dropdown(options=["none", "page_param", "cursor"], description="Pagination:")
119    api_page_param = w.Text(description="Page param:", value="page", disabled=True)
120    api_max_pages = w.IntText(description="Max pages:", value=20, disabled=True)
121    api_cursor_param = w.Text(description="Cursor param:", value="cursor", disabled=True)
122    api_cursor_json_path = w.Text(description="Cursor JSON path:", placeholder="meta.next_cursor", disabled=True)
123    api_pagination_box = w.VBox([])
124
125    def on_pagination_change(change):
126        for field in (api_page_param, api_max_pages, api_cursor_param, api_cursor_json_path):
127            field.disabled = True
128        if change["new"] == "page_param":
129            api_page_param.disabled = api_max_pages.disabled = False
130            api_pagination_box.children = [api_page_param, api_max_pages]
131        elif change["new"] == "cursor":
132            api_cursor_param.disabled = api_cursor_json_path.disabled = api_max_pages.disabled = False
133            api_pagination_box.children = [api_cursor_param, api_cursor_json_path, api_max_pages]
134        else:
135            api_pagination_box.children = []
136
137    api_pagination.observe(on_pagination_change, names="value")
138
139    api_headers_table = dashui.editable_table(["Header", "Value"], placeholders={"Header": "Accept", "Value": "application/json"})
140    api_params_table = dashui.editable_table(["Param", "Value"])
141
142    api_advanced = w.Accordion(children=[w.VBox([
143        api_auth_type, api_auth_box,
144        api_pagination, api_pagination_box,
145        w.HTML("<div style='font-size:12px;color:#5A6872;margin:6px 0 2px'>Headers</div>"),
146        api_headers_table.widget,
147        w.HTML("<div style='font-size:12px;color:#5A6872;margin:6px 0 2px'>Query params</div>"),
148        api_params_table.widget,
149    ])])
150    api_advanced.set_title(0, "Advanced")
151    api_advanced.selected_index = None
152
153    api_box = w.VBox([api_url, api_json_path, api_advanced])
154
155    # ── File format (path-based sources only) ───────────────────────────────
156    file_format = w.Dropdown(
157        options=["(infer from path)", "csv", "json", "parquet", "excel", "avro", "orc", "text"],
158        description="Format:",
159    )
160
161    csv_delimiter = w.Text(description="Delimiter:", value=",")
162    csv_header = w.Checkbox(value=True, description="Has header row")
163    csv_null_value = w.Text(description="Null marker:", placeholder="e.g. NA (optional)")
164    csv_box = w.VBox([w.HBox([csv_delimiter, csv_header]), csv_null_value])
165
166    # Excel options — the format that actually needs this much configuration
167    xl_sheet = w.Text(description="Sheet:", value="0", placeholder="name or 0-based index")
168    xl_header_row = w.IntText(description="Header row:", value=0, min=0)
169    xl_header = w.Checkbox(value=True, description="Has header row")
170    xl_password = w.Password(description="Password:", placeholder="if protected (optional)")
171    xl_sheets_to_union = w.Text(description="Union sheets:", placeholder="Jan, Feb, Mar (optional — reads+stacks multiple sheets)")
172    excel_box = w.VBox([
173        w.HBox([xl_sheet, xl_header_row, xl_header]),
174        xl_password, xl_sheets_to_union,
175    ])
176
177    format_options_panel = w.VBox([])
178
179    def on_format_change(change):
180        format_options_panel.children = {"csv": [csv_box], "excel": [excel_box]}.get(change["new"], [])
181
182    file_format.observe(on_format_change, names="value")
183
184    source_panel = w.VBox([vol_box])
185    format_row = w.VBox([file_format, format_options_panel])
186
187    def on_kind_change(change):
188        kind = change["new"]
189        source_panel.children = {
190            "Databricks Volume": [vol_box],
191            "ADLS Gen2": [adls_box],
192            "Amazon S3": [s3_box],
193            "DBFS": [dbfs_box],
194            "Database": [db_box],
195            "REST API": [api_box],
196        }[kind]
197        format_row.children = [] if kind in ("Database", "REST API") else [file_format, format_options_panel]
198
199    kind_toggle.observe(on_kind_change, names="value")
200    on_kind_change({"new": kind_toggle.value})
201
202    # ── Target ───────────────────────────────────────────────────────────────
203    target_table = w.Text(description="Target table:", placeholder="catalog.schema.table")
204    write_mode = w.ToggleButtons(options=["append", "overwrite", "merge"], description="Write mode:")
205    merge_keys = w.Text(description="Merge keys:", placeholder="id, updated_at (merge mode only)", disabled=True)
206    schema_evo = w.Checkbox(value=True, description="Allow schema evolution")
207    write_mode.observe(lambda c: setattr(merge_keys, "disabled", c["new"] != "merge"), names="value")
208
209    test_btn = dashui.action_button("Test Connection", style="info")
210    preview_btn = dashui.action_button("Preview", style="info")
211    run_btn = dashui.action_button("Run Ingestion", style="success")
212    output = dashui.output_panel()
213
214    def _build_reader_options(fmt):
215        from dashingest.readers import CsvReaderOptions, ExcelReaderOptions
216
217        if fmt == "csv":
218            return CsvReaderOptions(
219                delimiter=csv_delimiter.value or ",",
220                header=csv_header.value,
221                null_value=csv_null_value.value.strip(),
222            )
223        if fmt == "excel":
224            sheets = [s.strip() for s in xl_sheets_to_union.value.split(",") if s.strip()]
225            return ExcelReaderOptions(
226                sheet_name=xl_sheet.value.strip() or "0",
227                header_row=xl_header_row.value,
228                header=xl_header.value,
229                workbook_password=xl_password.value,
230                sheet_names=sheets,
231            )
232        return None
233
234    def _build_source():
235        from dashingest.connectors import ADLSSource, DatabaseSource, DBFSSource, RestApiSource, S3Source, VolumeSource
236
237        kind = kind_toggle.value
238        fmt = None if file_format.value == "(infer from path)" else file_format.value
239        reader_opts = _build_reader_options(fmt)
240
241        if kind == "Databricks Volume":
242            return VolumeSource(vol_catalog.value.strip(), vol_schema.value.strip(), vol_volume.value.strip(),
243                                 vol_path.value.strip(), fmt, reader_opts)
244        if kind == "ADLS Gen2":
245            return ADLSSource(adls_account.value.strip(), adls_container.value.strip(), adls_path.value.strip(),
246                               fmt, reader_opts)
247        if kind == "Amazon S3":
248            return S3Source(s3_bucket.value.strip(), s3_path.value.strip(), fmt, reader_opts)
249        if kind == "DBFS":
250            return DBFSSource(dbfs_path.value.strip(), fmt, reader_opts)
251        if kind == "Database":
252            port = int(db_port.value.strip()) if db_port.value.strip() else None
253            props = {r["Property"]: r["Value"] for r in db_props_table.values()}
254            return DatabaseSource(
255                host=db_host.value.strip(), database=db_database.value.strip(), engine=db_engine.value,
256                port=port, table=db_table.value.strip(), query=db_query.value.strip(),
257                user=db_user.value.strip(), password=db_password.value,
258                ssl=db_ssl.value,
259                fetch_size=db_fetch_size.value or None,
260                num_partitions=db_num_partitions.value or None,
261                partition_column=db_partition_col.value.strip(),
262                lower_bound=db_lower_bound.value if db_num_partitions.value else None,
263                upper_bound=db_upper_bound.value if db_num_partitions.value else None,
264                connection_properties=props,
265            )
266
267        headers = {r["Header"]: r["Value"] for r in api_headers_table.values()}
268        params = {r["Param"]: r["Value"] for r in api_params_table.values()}
269        return RestApiSource(
270            api_url.value.strip(), headers=headers, params=params, json_path=api_json_path.value.strip(),
271            auth_type=api_auth_type.value,
272            bearer_token=api_bearer_token.value,
273            api_key_header=api_key_header.value.strip() or "X-API-Key",
274            api_key=api_key_value.value,
275            basic_user=api_basic_user.value.strip(),
276            basic_password=api_basic_password.value,
277            pagination=api_pagination.value,
278            page_param=api_page_param.value.strip() or "page",
279            max_pages=api_max_pages.value or 20,
280            cursor_param=api_cursor_param.value.strip() or "cursor",
281            cursor_json_path=api_cursor_json_path.value.strip(),
282        )
283
284    # ── Config persistence — structural fields only, never passwords/tokens ──
285    def _collect_state() -> dict:
286        return {
287            "kind": kind_toggle.value,
288            "format": file_format.value,
289            "volume": {"catalog": vol_catalog.value, "schema": vol_schema.value, "volume": vol_volume.value, "path": vol_path.value},
290            "adls": {"account": adls_account.value, "container": adls_container.value, "path": adls_path.value},
291            "s3": {"bucket": s3_bucket.value, "path": s3_path.value},
292            "dbfs": {"path": dbfs_path.value},
293            "database": {
294                "engine": db_engine.value, "host": db_host.value, "database": db_database.value,
295                "port": db_port.value, "table": db_table.value, "query": db_query.value, "user": db_user.value,
296                "ssl": db_ssl.value,
297            },
298            "rest_api": {"url": api_url.value, "json_path": api_json_path.value, "auth_type": api_auth_type.value, "pagination": api_pagination.value},
299            "csv": {"delimiter": csv_delimiter.value, "header": csv_header.value, "null_value": csv_null_value.value},
300            "excel": {"sheet": xl_sheet.value, "header_row": xl_header_row.value, "header": xl_header.value},
301            "target": {
302                "table": target_table.value, "write_mode": write_mode.value,
303                "merge_keys": merge_keys.value, "schema_evolution": schema_evo.value,
304            },
305        }
306
307    def _apply_state(state: dict) -> None:
308        if not state:
309            return
310        kind_toggle.value = state.get("kind", kind_toggle.value)
311        file_format.value = state.get("format", file_format.value)
312        v = state.get("volume", {})
313        vol_catalog.value, vol_schema.value, vol_volume.value, vol_path.value = (
314            v.get("catalog", ""), v.get("schema", ""), v.get("volume", ""), v.get("path", ""))
315        a = state.get("adls", {})
316        adls_account.value, adls_container.value, adls_path.value = (
317            a.get("account", ""), a.get("container", ""), a.get("path", ""))
318        s3 = state.get("s3", {})
319        s3_bucket.value, s3_path.value = s3.get("bucket", ""), s3.get("path", "")
320        dbfs_path.value = state.get("dbfs", {}).get("path", "")
321        d = state.get("database", {})
322        db_engine.value = d.get("engine", db_engine.value)
323        db_host.value, db_database.value, db_port.value = d.get("host", ""), d.get("database", ""), d.get("port", "")
324        db_table.value, db_query.value, db_user.value = d.get("table", ""), d.get("query", ""), d.get("user", "")
325        db_ssl.value = d.get("ssl", False)
326        r = state.get("rest_api", {})
327        api_url.value, api_json_path.value = r.get("url", ""), r.get("json_path", "")
328        api_auth_type.value = r.get("auth_type", api_auth_type.value)
329        api_pagination.value = r.get("pagination", api_pagination.value)
330        c = state.get("csv", {})
331        csv_delimiter.value = c.get("delimiter", ",")
332        csv_header.value = c.get("header", True)
333        csv_null_value.value = c.get("null_value", "")
334        e = state.get("excel", {})
335        xl_sheet.value = e.get("sheet", "0")
336        xl_header_row.value = e.get("header_row", 0)
337        xl_header.value = e.get("header", True)
338        t = state.get("target", {})
339        target_table.value = t.get("table", "")
340        write_mode.value = t.get("write_mode", write_mode.value)
341        merge_keys.value = t.get("merge_keys", "")
342        schema_evo.value = t.get("schema_evolution", True)
343
344    def _save_state() -> None:
345        try:
346            dashui.save_config(_LIBRARY, _collect_state())
347        except Exception:
348            pass  # persistence is a convenience, never block the actual operation on it
349
350    _apply_state(dashui.load_config(_LIBRARY))
351
352    def on_test(b):
353        with output:
354            output.clear_output()
355            try:
356                from dashingest.ingestor import test_connection
357                source = _build_source()
358                _save_state()
359                test_connection(source).display()
360            except Exception as e:
361                print(f"Error: {e}")
362
363    def on_preview(b):
364        with output:
365            output.clear_output()
366            try:
367                from dashingest.ingestor import preview
368                source = _build_source()
369                _save_state()
370                print(preview(source, limit=10))
371            except Exception as e:
372                print(f"Error: {e}")
373
374    def on_run(b):
375        with output:
376            output.clear_output()
377            try:
378                from dashingest.connectors import IngestTarget
379                from dashingest.ingestor import run_ingestion
380
381                target = IngestTarget(
382                    table=target_table.value.strip(),
383                    write_mode=write_mode.value,
384                    schema_evolution=schema_evo.value,
385                    merge_keys=[k.strip() for k in merge_keys.value.split(",") if k.strip()],
386                )
387                source = _build_source()
388                _save_state()
389                result = run_ingestion(source, target)
390                result.display()
391            except Exception as e:
392                print(f"Error: {e}")
393
394    test_btn.on_click(on_test)
395    preview_btn.on_click(on_preview)
396    run_btn.on_click(on_run)
397
398    env_accordion = w.Accordion(children=[dashui.env_setup_panel(_LIBRARY).widget])
399    env_accordion.set_title(0, "Environment setup")
400    env_accordion.selected_index = None
401
402    ui = dashui.card([
403        dashui.header("DashIngest — Data Ingestion", library="dashingest"),
404        env_accordion,
405        dashui.section("Step 1: Source"),
406        kind_toggle, source_panel, format_row,
407        w.HBox([test_btn, preview_btn]),
408        dashui.section("Step 2: Target"),
409        target_table, write_mode, merge_keys, schema_evo,
410        dashui.section("Step 3: Run"),
411        run_btn, output,
412    ])
413    display(ui)