File size: 4,625 Bytes
eed1cab 791c076 eed1cab | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """CLI subcommand: ``dataforge watch``."""
from __future__ import annotations
import json
import time
from pathlib import Path
from typing import Annotated, Literal
import typer
from rich.console import Console
from rich.panel import Panel
from dataforge.cli.common import load_schema, read_csv, resolve_cli_path
from dataforge.detectors import run_all_detectors
from dataforge.detectors.base import Schema
from dataforge.ui.profile_view import render_profile_table
from dataforge.ui.repair_diff import render_repair_diff
_console = Console(stderr=True)
WatchAction = Literal["profile", "repair"]
def _load_optional_schema(schema_path: Path | None) -> Schema | None:
if schema_path is None:
return None
resolved_schema = resolve_cli_path(schema_path)
if not resolved_schema.exists():
raise typer.BadParameter(f"Schema file '{schema_path}' does not exist.")
return load_schema(resolved_schema)
def _profile_once(path: Path, schema: Schema | None, json_output: bool) -> None:
df = read_csv(path)
issues = run_all_detectors(df, schema)
if json_output:
typer.echo(
json.dumps(
{
"event": "profile",
"path": str(path),
"issues_count": len(issues),
"issues": [issue.model_dump(mode="json") for issue in issues],
},
indent=2,
sort_keys=True,
)
)
return
render_profile_table(issues, Console(), file_path=str(path))
def _repair_once(path: Path, schema: Schema | None, apply: bool, json_output: bool) -> None:
from dataforge.engine.repair import RepairPipelineRequest, run_repair_pipeline
result = run_repair_pipeline(
RepairPipelineRequest(
source_path=path,
mode="apply" if apply else "dry_run",
schema=schema,
interactive=False,
)
)
if json_output:
payload = result.model_dump(mode="json")
payload["event"] = "repair"
typer.echo(json.dumps(payload, indent=2, sort_keys=True))
return
render_repair_diff(result.fixes, Console(), file_path=str(path))
def _run_once(
path: Path, schema: Schema | None, action: WatchAction, apply: bool, json: bool
) -> None:
if action == "repair":
_repair_once(path, schema, apply, json)
else:
_profile_once(path, schema, json)
def watch(
path: Annotated[
Path,
typer.Argument(help="CSV or dbt artifact path to watch."),
],
schema: Annotated[
Path | None,
typer.Option("--schema", help="Path to a YAML schema file with column types and FDs."),
] = None,
action: Annotated[
WatchAction,
typer.Option("--action", help="Action to run when the file changes: profile or repair."),
] = "profile",
apply: Annotated[
bool,
typer.Option("--apply", help="Apply repairs on change. Defaults to dry-run repair."),
] = False,
interval: Annotated[
float,
typer.Option("--interval", min=0.1, help="Polling interval in seconds."),
] = 2.0,
once: Annotated[
bool,
typer.Option("--once", help="Run once and exit, useful for CI acceptance."),
] = False,
json_output: Annotated[
bool,
typer.Option("--json", help="Print watch events as JSON."),
] = False,
) -> None:
"""Poll a path and rerun profile or repair when it changes."""
resolved_path = resolve_cli_path(path)
if not resolved_path.exists():
_console.print(f"[bold red]Watch path not found:[/bold red] {path}")
raise typer.Exit(code=2)
parsed_schema = _load_optional_schema(schema)
if apply and action != "repair":
_console.print(
Panel(
"--apply is only valid with --action repair.",
title="Watch Error",
style="red",
)
)
raise typer.Exit(code=2)
_run_once(resolved_path, parsed_schema, action, apply, json_output)
if once:
return
last_mtime = resolved_path.stat().st_mtime_ns
while True:
time.sleep(interval)
try:
current_mtime = resolved_path.stat().st_mtime_ns
except FileNotFoundError:
_console.print(f"[bold red]Watch path disappeared:[/bold red] {resolved_path}")
raise typer.Exit(code=2) from None
if current_mtime == last_mtime:
continue
last_mtime = current_mtime
_run_once(resolved_path, parsed_schema, action, apply, json_output)
|