199 lines
7.1 KiB
Python
199 lines
7.1 KiB
Python
from pathlib import Path
|
|
|
|
import typer
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
from tht.cli.config_cmd import CONFIG_OPT
|
|
from tht.config import ConfigError, load_config
|
|
from tht.db.connection import make_engine
|
|
from tht.db.introspect import IntrospectionError, introspect
|
|
from tht.db.sampling import add_examples
|
|
from tht.mschema.eligibility import classify_all
|
|
|
|
schema_app = typer.Typer(help="Gestione mschema (rappresentazione canonica dello schema)")
|
|
|
|
|
|
def _load_config_or_exit(config: Path):
|
|
try:
|
|
return load_config(config)
|
|
except ConfigError as e:
|
|
typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
|
|
|
|
def physical_path(cfg) -> Path:
|
|
return cfg.paths.artifacts / "mschema" / "physical.yaml"
|
|
|
|
|
|
def annotations_path(cfg) -> Path:
|
|
return cfg.paths.artifacts / "mschema" / "annotations.yaml"
|
|
|
|
|
|
@schema_app.command("introspect")
|
|
def introspect_cmd(config: Path = CONFIG_OPT) -> None:
|
|
"""Introspeziona lo schema target e genera artifacts/mschema/physical.yaml."""
|
|
cfg = _load_config_or_exit(config)
|
|
if cfg.database.transport == "rest":
|
|
from tht.db.introspect import introspect_rest
|
|
from tht.db.sampling import add_examples_rest
|
|
from tht.rest.client import RestClient, RestError
|
|
|
|
client = RestClient(cfg.rest)
|
|
try:
|
|
phys = introspect_rest(
|
|
client, database=cfg.database.database, schema=cfg.database.db_schema
|
|
)
|
|
add_examples_rest(client, phys, cfg.examples)
|
|
classify_all(phys, cfg.eligibility)
|
|
except RestError as e:
|
|
typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
else:
|
|
engine = make_engine(cfg.database)
|
|
try:
|
|
phys = introspect(
|
|
engine, database=cfg.database.database, schema=cfg.database.db_schema
|
|
)
|
|
add_examples(engine, phys, cfg.examples)
|
|
classify_all(phys, cfg.eligibility)
|
|
except (OperationalError, IntrospectionError) as e:
|
|
typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
out = physical_path(cfg)
|
|
phys.to_yaml(out)
|
|
n_cols = sum(len(t.columns) for t in phys.tables.values())
|
|
n_ignored = sum(
|
|
1 for t in phys.tables.values() for c in t.columns.values() if not c.eligible
|
|
)
|
|
typer.secho(
|
|
f"OK: {len(phys.tables)} tabelle, {n_cols} colonne "
|
|
f"({n_ignored} ignorate: testo ampio) -> {out}",
|
|
fg=typer.colors.GREEN,
|
|
)
|
|
|
|
|
|
@schema_app.command("check")
|
|
def check_cmd(config: Path = CONFIG_OPT) -> None:
|
|
"""Confronta physical.yaml e annotations.yaml; segnala annotazioni orfane."""
|
|
from tht.mschema.merge import find_orphans
|
|
from tht.mschema.models import Annotations, PhysicalSchema
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
phys_file = physical_path(cfg)
|
|
if not phys_file.exists():
|
|
typer.secho(
|
|
f"ERRORE: {phys_file} non trovato. Esegui prima `tht schema introspect`.",
|
|
fg=typer.colors.RED, err=True,
|
|
)
|
|
raise typer.Exit(code=1)
|
|
physical = PhysicalSchema.from_yaml(phys_file)
|
|
annotations = Annotations.from_yaml(annotations_path(cfg))
|
|
|
|
ignored = [
|
|
f"{t}.{c} ({col.eligibility_reason})"
|
|
for t, table in physical.tables.items()
|
|
for c, col in table.columns.items()
|
|
if not col.eligible
|
|
]
|
|
if ignored:
|
|
typer.secho(
|
|
f"Colonne ignorate (testo ampio, {len(ignored)}):", fg=typer.colors.YELLOW
|
|
)
|
|
for line in ignored:
|
|
typer.echo(f" - {line}")
|
|
|
|
orphans = find_orphans(physical, annotations)
|
|
if orphans:
|
|
typer.secho(f"ATTENZIONE: {len(orphans)} annotazioni orfane:", fg=typer.colors.YELLOW)
|
|
for o in orphans:
|
|
typer.echo(f" - {o}")
|
|
raise typer.Exit(code=3)
|
|
typer.secho("OK: nessuna annotazione orfana.", fg=typer.colors.GREEN)
|
|
|
|
|
|
@schema_app.command("render")
|
|
def render_cmd(
|
|
config: Path = CONFIG_OPT,
|
|
format: str = typer.Option(
|
|
"markdown", "--format", "-f", help="Formato: markdown | mschema-text | schema-dict"
|
|
),
|
|
tables: list[str] = typer.Option(
|
|
None, "--table", "-t", help="Limita alle tabelle indicate (ripetibile)."
|
|
),
|
|
output: Path = typer.Option(None, "--output", "-o", help="File di output (default stdout)."),
|
|
) -> None:
|
|
"""Serializza mschema (physical + annotations) nel formato richiesto."""
|
|
import json
|
|
|
|
from tht.mschema.models import Annotations, PhysicalSchema
|
|
from tht.mschema.render import to_markdown, to_mschema_text, to_schema_dict
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
phys_file = physical_path(cfg)
|
|
if not phys_file.exists():
|
|
typer.secho(
|
|
f"ERRORE: {phys_file} non trovato. Esegui prima `tht schema introspect`.",
|
|
fg=typer.colors.RED, err=True,
|
|
)
|
|
raise typer.Exit(code=1)
|
|
physical = PhysicalSchema.from_yaml(phys_file)
|
|
annotations = Annotations.from_yaml(annotations_path(cfg))
|
|
table_filter = list(tables) if tables else None
|
|
|
|
if format == "markdown":
|
|
out = to_markdown(physical, annotations)
|
|
elif format == "mschema-text":
|
|
out = to_mschema_text(physical, annotations, tables=table_filter)
|
|
elif format == "schema-dict":
|
|
out = json.dumps(to_schema_dict(physical, annotations), ensure_ascii=False, indent=2)
|
|
else:
|
|
typer.secho(f"ERRORE: formato sconosciuto: {format}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
|
|
if output:
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
output.write_text(out)
|
|
typer.secho(f"OK: scritto {output}", fg=typer.colors.GREEN)
|
|
else:
|
|
typer.echo(out)
|
|
|
|
|
|
@schema_app.command("columns")
|
|
def columns_cmd(
|
|
table: str = typer.Argument(..., help="Nome tabella (chiave in physical.yaml)."),
|
|
json_out: bool = typer.Option(False, "--json", help="Emetti JSON puro su stdout."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Elenca nome/descrizione/tipo/pk delle colonne di una tabella dal catalogo."""
|
|
import json as _json
|
|
|
|
from tht.mschema.models import PhysicalSchema
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
phys_file = physical_path(cfg)
|
|
if not phys_file.exists():
|
|
typer.secho(
|
|
f"ERRORE: {phys_file} non trovato. Esegui prima `tht schema introspect`.",
|
|
fg=typer.colors.RED, err=True,
|
|
)
|
|
raise typer.Exit(code=1)
|
|
physical = PhysicalSchema.from_yaml(phys_file)
|
|
tbl = physical.tables.get(table)
|
|
if tbl is None:
|
|
typer.secho(f"ERRORE: tabella non nel catalogo: {table}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
payload = {
|
|
"table": table,
|
|
"description": tbl.comment,
|
|
"columns": [
|
|
{"name": name, "description": col.comment, "type": col.type, "pk": col.pk}
|
|
for name, col in tbl.columns.items()
|
|
],
|
|
}
|
|
if json_out:
|
|
typer.echo(_json.dumps(payload, ensure_ascii=False))
|
|
return
|
|
typer.echo(f"{table}: {tbl.comment}")
|
|
for c in payload["columns"]:
|
|
typer.echo(f" {'*' if c['pk'] else ' '} {c['name']} ({c['type']}) — {c['description']}")
|