File size: 4,918 Bytes
5143557 eed1cab 5143557 eed1cab 791c076 5143557 eed1cab 5143557 eed1cab 5143557 eed1cab 791c076 5143557 eed1cab 5143557 eed1cab 791c076 eed1cab 5143557 eed1cab 5143557 eed1cab 5143557 eed1cab 5143557 eed1cab 5143557 791c076 5143557 eed1cab 791c076 eed1cab 5143557 eed1cab 5143557 | 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 146 147 148 | """CLI subcommand: ``dataforge profile <path> [--schema <yaml>]``.
Reads a CSV file, runs all detectors, and renders detected issues as a
rich-formatted terminal table. Diagnostics exit 0 by default; use
``--fail-on`` for CI gating.
"""
from __future__ import annotations
import json
from collections.abc import Sequence
from hashlib import sha256
from pathlib import Path
from typing import Annotated, Literal
import typer
from rich.console import Console
from dataforge.cli.common import load_schema, read_csv, resolve_cli_path
from dataforge.detectors import run_all_detectors
from dataforge.detectors.base import Issue, Schema, Severity
from dataforge.schema_inference import (
build_constraint_review_artifact,
dump_constraint_review_artifact,
infer_schema,
)
from dataforge.ui.profile_view import render_profile_table
_console = Console(stderr=True)
FailOn = Literal["never", "unsafe", "review", "any"]
def _should_fail(issues: Sequence[Issue], fail_on: FailOn) -> bool:
"""Return whether profile findings should trip the requested CI gate."""
if fail_on == "never":
return False
if fail_on == "any":
return bool(issues)
severities = [issue.severity for issue in issues]
if fail_on == "unsafe":
return any(severity == Severity.UNSAFE for severity in severities)
return any(severity >= Severity.REVIEW for severity in severities)
def profile(
path: Annotated[
Path,
typer.Argument(
help="Path to the CSV file to profile.",
),
],
schema: Annotated[
Path | None,
typer.Option(
"--schema",
help="Path to a YAML schema file with column types and FDs.",
),
] = None,
json_output: Annotated[
bool,
typer.Option("--json", help="Print profile results as JSON."),
] = False,
constraints_out: Annotated[
Path | None,
typer.Option(
"--constraints-out",
help="Write inferred constraints as a pending review artifact.",
),
] = None,
fail_on: Annotated[
FailOn,
typer.Option(
"--fail-on",
help="Exit 1 when findings meet this threshold: never, unsafe, review, any.",
),
] = "never",
) -> None:
"""Profile a CSV file for data-quality issues.
Reads the CSV, runs all detectors (type_mismatch, decimal_shift,
fd_violation), and renders a rich-formatted table of detected issues.
Exit code 0 unless ``--fail-on`` is set and matching findings are present.
"""
resolved_path = resolve_cli_path(path)
if not resolved_path.exists():
_console.print(f"[bold red]CSV file not found:[/bold red] {path}")
raise typer.Exit(code=2)
# Load the CSV with dtype=str to avoid pandas type-coercion artifacts.
try:
df = read_csv(resolved_path)
except Exception as exc:
_console.print(f"[bold red]Error reading CSV:[/bold red] {exc}")
raise typer.Exit(code=2) from exc
# Optionally load schema.
parsed_schema: Schema | None = None
if schema is not None:
resolved_schema = resolve_cli_path(schema)
if not resolved_schema.exists():
_console.print(f"[bold red]Schema file not found:[/bold red] {schema}")
raise typer.Exit(code=2)
parsed_schema = load_schema(resolved_schema)
# Run all detectors.
issues = run_all_detectors(df, parsed_schema)
schema_inference = infer_schema(df)
source_sha256 = sha256(resolved_path.read_bytes()).hexdigest()
if constraints_out is not None:
try:
artifact = build_constraint_review_artifact(
schema_inference,
source_path=resolved_path,
source_sha256=source_sha256,
)
constraints_out.parent.mkdir(parents=True, exist_ok=True)
constraints_out.write_text(
dump_constraint_review_artifact(artifact),
encoding="utf-8",
)
except Exception as exc:
_console.print(f"[bold red]Error writing constraints artifact:[/bold red] {exc}")
raise typer.Exit(code=2) from exc
# Render the results.
if json_output:
typer.echo(
json.dumps(
{
"path": str(resolved_path),
"issues_count": len(issues),
"fail_on": fail_on,
"issues": [issue.model_dump(mode="json") for issue in issues],
"schema_inference": schema_inference.model_dump(mode="json"),
},
indent=2,
sort_keys=True,
)
)
else:
output_console = Console()
render_profile_table(issues, output_console, file_path=str(resolved_path))
if _should_fail(issues, fail_on):
raise typer.Exit(code=1)
|