diff --git a/harness/tests/test_adapter_factory.py b/harness/tests/test_adapter_factory.py new file mode 100644 index 00000000..cf0f2f4b --- /dev/null +++ b/harness/tests/test_adapter_factory.py @@ -0,0 +1,86 @@ +import pytest + +from tht.adapters.dwh import PostgresDwhAdapter, ThothRestDwhAdapter +from tht.adapters.vector import LegacyDirectVectorStore, ThothHttpVectorStore +from tht.adapters.factory import build_dwh, build_vector_store +from tht.config import Config, ConfigError + + +def _config(*, dwh_type="thoth_rest", vector_type="thoth_vector_http", writer=True): + dwh = ( + { + "type": "thoth_rest", + "database": {"database": "analytics", "schema": "mart"}, + "endpoint": {"base_url": "https://dwh.test/", "api_key": "reader"}, + } + if dwh_type == "thoth_rest" + else { + "type": "postgres_direct", + "connection": { + "host": "db", + "database": "analytics", + "schema": "mart", + "user": "reader", + "password": "secret", + }, + } + ) + vectors = ( + { + "type": "thoth_vector_http", + "reader": {"base_url": "https://vectors.test/", "api_key": "reader"}, + **( + {"writer": {"base_url": "https://vectors.test/", "api_key": "writer"}} + if writer + else {} + ), + } + if vector_type == "thoth_vector_http" + else { + "type": "pgvector_direct", + "connection": { + "host": "vector-db", + "database": "postgres", + "schema": "vectors", + "user": "reader", + "password": "secret", + }, + } + ) + legacy_database = ( + dwh["connection"] + if dwh_type == "postgres_direct" + else { + **dwh["database"], + "user": "rest", + "password": "", + "transport": "rest", + } + ) + return Config.model_validate( + {"dwh": dwh, "vectors": vectors, "database": legacy_database} + ) + + +@pytest.mark.parametrize( + ("dwh_type", "adapter_type"), + [("postgres_direct", PostgresDwhAdapter), ("thoth_rest", ThothRestDwhAdapter)], +) +def test_factory_selects_dwh_adapter(dwh_type, adapter_type): + assert isinstance(build_dwh(_config(dwh_type=dwh_type)), adapter_type) + + +def test_factory_selects_http_vector_and_requires_writer(): + config = _config(writer=False) + + assert isinstance(build_vector_store(config), ThothHttpVectorStore) + with pytest.raises(ConfigError, match="writer"): + build_vector_store(config, require_write=True) + + +def test_factory_selects_direct_vector_reader(): + config = _config(vector_type="pgvector_direct") + + assert isinstance(build_vector_store(config), LegacyDirectVectorStore) + with pytest.raises(ConfigError, match="writer"): + build_vector_store(config, require_write=True) diff --git a/harness/tht/adapters/factory.py b/harness/tht/adapters/factory.py new file mode 100644 index 00000000..61fffa60 --- /dev/null +++ b/harness/tht/adapters/factory.py @@ -0,0 +1,83 @@ +"""Central construction of deployment-specific adapters.""" + +from tht.adapters.dwh import PostgresDwhAdapter, ThothRestDwhAdapter +from tht.adapters.vector import LegacyDirectVectorStore, ThothHttpVectorStore +from tht.config import Config, ConfigError +from tht.db.connection import make_engine +from tht.ports.dwh import DwhAdapter +from tht.ports.vector import VectorStore +from tht.vectorstore.rest_client import VectorRestClient + + +def build_dwh(cfg: Config) -> DwhAdapter: + """Build the DWH adapter selected by the validated workspace resource.""" + resource = cfg.dwh + match resource.type: + case "postgres_direct": + return PostgresDwhAdapter(resource.connection) + case "thoth_rest": + return ThothRestDwhAdapter(resource.database, resource.endpoint) + case other: # pragma: no cover - Pydantic's discriminator rejects this first. + raise ConfigError(f"Adapter DWH non supportato: {other}") + + +def build_vector_store(cfg: Config, *, require_write: bool = False) -> VectorStore: + """Build the vector adapter, optionally requiring an HTTP writer credential.""" + resource = cfg.vectors + if resource is None: + raise ConfigError("Risorsa vectors non configurata") + + match resource.type: + case "pgvector_direct": + if require_write: + raise ConfigError("Vector writer non configurato per pgvector_direct") + dim = cfg.embeddings.dim if cfg.embeddings is not None else 768 + return LegacyDirectVectorStore( + make_engine(resource.connection), + schema=resource.connection.db_schema, + dim=dim, + ) + case "thoth_vector_http": + if resource.reader is None: + raise ConfigError("Vector reader non configurato") + if require_write and resource.writer is None: + raise ConfigError("Vector writer non configurato") + return ThothHttpVectorStore( + VectorRestClient(resource.reader), + VectorRestClient(resource.writer) if resource.writer is not None else None, + ) + case other: # pragma: no cover - Pydantic's discriminator rejects this first. + raise ConfigError(f"Adapter vector non supportato: {other}") + + +def build_vector_loader(cfg: Config, collection: str): + """Compatibility construction for legacy collection sync commands.""" + resource = cfg.vectors + if resource is None: + raise ConfigError("Risorsa vectors non configurata") + if cfg.embeddings is None: + raise ConfigError("Embeddings non configurati") + + if resource.type == "thoth_vector_http" and resource.writer is not None and ( + cfg.profile == "workstation" or resource.direct is None + ): + from tht.vectorstore.rest_writer import RestVectorWriter + + return RestVectorWriter(VectorRestClient(resource.writer), table=collection) + + connection = ( + resource.connection if resource.type == "pgvector_direct" else resource.direct + ) + if connection is None: + raise ConfigError("Vector writer non configurato") + from tht.vectorstore.store import VectorStore as TableVectorStore + + return TableVectorStore( + make_engine(connection), + schema=connection.db_schema, + table=collection, + dim=cfg.embeddings.dim, + ) + + +__all__ = ["build_dwh", "build_vector_loader", "build_vector_store"] diff --git a/harness/tht/cli/db_cmd.py b/harness/tht/cli/db_cmd.py index 549262d1..28ed309a 100644 --- a/harness/tht/cli/db_cmd.py +++ b/harness/tht/cli/db_cmd.py @@ -3,6 +3,8 @@ from pathlib import Path import typer from sqlalchemy.exc import OperationalError +from tht.adapters.dwh import ThothRestDwhAdapter +from tht.adapters.factory import build_dwh from tht.cli.config_cmd import CONFIG_OPT from tht.config import ConfigError, load_config from tht.db.connection import can_create_in_schema, make_engine, ping, writable_tables @@ -11,20 +13,21 @@ from tht.db.fetch_ca import CaFetchError, describe_pem, fetch_chain_pem, parse_h db_app = typer.Typer(help="Operazioni sul database target") -def _ping_rest(cfg, schema: str) -> None: +def _ping_rest(adapter: ThothRestDwhAdapter, cfg, schema: str) -> None: """Health check via REST. Il read-only è garantito strutturalmente dall'API (ammette solo SELECT/WITH): non serve il controllo dei privilegi di scrittura.""" - from tht.rest.client import RestClient, RestError + from tht.rest.client import RestError try: - info = RestClient(cfg.rest).ping() + info = adapter._client.ping() except RestError as e: typer.secho(f"ERRORE di connessione: {e}", fg=typer.colors.RED, err=True) raise typer.Exit(code=1) if not info.get("db_connected") or not info.get("schema_accessible"): typer.secho( f"ERRORE: DWH non accessibile via REST (risposta: {info}).", - fg=typer.colors.RED, err=True, + fg=typer.colors.RED, + err=True, ) raise typer.Exit(code=1) typer.secho( @@ -44,8 +47,9 @@ def ping_cmd(config: Path = CONFIG_OPT) -> None: typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True) raise typer.Exit(code=1) schema = cfg.database.db_schema - if cfg.database.transport == "rest": - _ping_rest(cfg, schema) + adapter = build_dwh(cfg) + if isinstance(adapter, ThothRestDwhAdapter): + _ping_rest(adapter, cfg, schema) return engine = make_engine(cfg.database) try: diff --git a/harness/tht/cli/lsh_cmd.py b/harness/tht/cli/lsh_cmd.py index c55c90af..a275633f 100644 --- a/harness/tht/cli/lsh_cmd.py +++ b/harness/tht/cli/lsh_cmd.py @@ -32,20 +32,30 @@ def build_cmd(config: Path = CONFIG_OPT) -> None: annotations = Annotations.from_yaml(annotations_path(cfg)) typer.echo("Estrazione valori (i più frequenti) dalle colonne testuali eligible...") - if cfg.database.transport == "rest": - from tht.db.sampling import unique_values_for_lsh_rest - from tht.rest.client import RestClient + from tht.adapters.factory import build_dwh + from tht.db.sampling import SkippedColumn, TruncatedColumn, is_text_type + from tht.mschema.eligibility import effective_eligibility - values, skipped, truncated = unique_values_for_lsh_rest( - RestClient(cfg.rest), physical, cfg.lsh, annotations - ) - else: - from tht.db.connection import make_engine - from tht.db.sampling import unique_values_for_lsh - - values, skipped, truncated = unique_values_for_lsh( - make_engine(cfg.database), physical, cfg.lsh, annotations - ) + dwh = build_dwh(cfg) + values: dict[str, dict[str, list[str]]] = {} + skipped: list[SkippedColumn] = [] + truncated: list[TruncatedColumn] = [] + for table_name, table in physical.tables.items(): + table_ann = annotations.tables.get(table_name) + for column_name, column in table.columns.items(): + ann_col = table_ann.columns.get(column_name) if table_ann else None + if not is_text_type(column.type) or not effective_eligibility(column, ann_col)[0]: + continue + try: + distinct = dwh.distinct_values(table_name, column_name) + except Exception as e: + skipped.append(SkippedColumn(table_name, column_name, f"errore: {e}")) + continue + vals = [str(value) for value in distinct.values if value not in (None, "")] + if vals: + values.setdefault(table_name, {})[column_name] = vals + if distinct.truncated or len(vals) >= cfg.lsh.max_values_per_column: + truncated.append(TruncatedColumn(table_name, column_name, len(vals))) n_values = sum(len(v) for t in values.values() for v in t.values()) typer.echo(f" {n_values} valori da {sum(len(t) for t in values.values())} colonne") for s in skipped: diff --git a/harness/tht/cli/memory_cmd.py b/harness/tht/cli/memory_cmd.py index f0ee7571..9ba16ca6 100644 --- a/harness/tht/cli/memory_cmd.py +++ b/harness/tht/cli/memory_cmd.py @@ -153,6 +153,7 @@ def save_one_cmd( """ import json as _json + from tht.adapters.factory import build_vector_store from tht.cli.vector_cmd import make_embedder from tht.memory import load_registry, promote, save_one_memory from tht.vectorstore.rest_client import VectorRestClient @@ -167,6 +168,7 @@ def save_one_cmd( fg=typer.colors.RED, err=True, ) raise typer.Exit(code=4) + build_vector_store(cfg, require_write=True) sdir = session_dir(cfg, session) # Promuove la decisione scelta nel registro locale (idempotente: salta se gia' presente @@ -449,6 +451,7 @@ def index_solved_session(cfg, session_id: str) -> int: Solleva RuntimeError se manca la writer key e SolvedIndexError se mancano gli artefatti: il finalize li degrada a warning, il comando CLI li converte in errori espliciti.""" + from tht.adapters.factory import build_vector_store from tht.cli.sql_cmd import promoted_tables_for from tht.cli.vector_cmd import make_embedder from tht.solved import build_solved_record, save_solved_question @@ -459,6 +462,7 @@ def index_solved_session(cfg, session_id: str) -> int: "vector_write_rest assente: la coppia domanda->SQL si indicizza solo con la " "writer key configurata nel workspace yaml" ) + build_vector_store(cfg, require_write=True) manifest = load_session_or_exit(cfg, session_id) record = build_solved_record( session_dir(cfg, session_id), manifest, promoted_tables_for(cfg, session_id) diff --git a/harness/tht/cli/schema_cmd.py b/harness/tht/cli/schema_cmd.py index e2e671a1..f8fe23aa 100644 --- a/harness/tht/cli/schema_cmd.py +++ b/harness/tht/cli/schema_cmd.py @@ -1,13 +1,10 @@ from pathlib import Path import typer -from sqlalchemy.exc import OperationalError - +from tht.adapters.factory import build_dwh 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.db.sampling import is_text_type from tht.mschema.eligibility import classify_all schema_app = typer.Typer(help="Gestione mschema (rappresentazione canonica dello schema)") @@ -64,32 +61,25 @@ def introspect_cmd( fg=typer.colors.GREEN, ) return - 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) + try: + dwh = build_dwh(cfg) + phys = dwh.introspect() + for table_name, table in phys.tables.items(): + for column_name, column in table.columns.items(): + if is_text_type(column.type): + column.examples = [ + str(value) + for value in dwh.sample_column( + table_name, + column_name, + limit=cfg.examples.max_per_column, + ) + if value not in (None, "") + ] + classify_all(phys, cfg.eligibility) + except Exception as e: + typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True) + raise typer.Exit(code=1) phys.to_yaml(out) n_cols = sum(len(t.columns) for t in phys.tables.values()) n_ignored = sum( diff --git a/harness/tht/cli/sql_cmd.py b/harness/tht/cli/sql_cmd.py index 5fd1a0d3..e1dc609b 100644 --- a/harness/tht/cli/sql_cmd.py +++ b/harness/tht/cli/sql_cmd.py @@ -90,46 +90,18 @@ def validate_or_exit(cfg, sql: str, session_id: str | None): return result -def _ro_engine(cfg): - """Engine sul target con search_path impostato allo schema (nomi non qualificati).""" - from sqlalchemy import create_engine - - db = cfg.database - url = f"postgresql+psycopg2://{db.user}:{db.password}@{db.host}:{db.port}/{db.database}" - return create_engine( - url, echo=False, - connect_args={"options": f"-csearch_path={db.db_schema}"}, - ) - - -def _rest_client(cfg): - from tht.rest.client import RestClient - - return RestClient(cfg.rest) - - def do_explain(cfg, sql: str): - """EXPLAIN secondo il transport configurato (direct|rest).""" - if cfg.database.transport == "rest": - from tht.rest.execute import explain_rest + """EXPLAIN through the configured DWH adapter.""" + from tht.adapters.factory import build_dwh - return explain_rest(_rest_client(cfg), sql) - from tht.execute import explain - - return explain(_ro_engine(cfg), sql, timeout_ms=cfg.execution.statement_timeout_ms) + return build_dwh(cfg).explain(sql) def _run_transport(cfg, sql: str, *, limit: int): - """Dispatch all'esecutore controllato secondo il transport (direct|rest).""" - if cfg.database.transport == "rest": - from tht.rest.execute import run_controlled_rest + """Dispatch through the configured DWH adapter.""" + from tht.adapters.factory import build_dwh - return run_controlled_rest(_rest_client(cfg), sql, limit=limit) - from tht.execute import run_controlled - - return run_controlled( - _ro_engine(cfg), sql, limit=limit, timeout_ms=cfg.execution.statement_timeout_ms - ) + return build_dwh(cfg).run_query(sql, limit=limit) def do_run(cfg, sql: str, *, limit: int, offset: int = 0): diff --git a/harness/tht/cli/vector_cmd.py b/harness/tht/cli/vector_cmd.py index 102cf0f9..7a297876 100644 --- a/harness/tht/cli/vector_cmd.py +++ b/harness/tht/cli/vector_cmd.py @@ -46,35 +46,26 @@ def open_store(cfg, table: str): Sul server preferisce la connessione diretta. In profilo workstation usa `vector_write_rest` se configurato, con upsert remoto non distruttivo. """ - if has_vector_write_rest(cfg) and (cfg.profile == "workstation" or cfg.vector_db is None): - from tht.vectorstore.rest_client import VectorRestClient - from tht.vectorstore.rest_writer import RestVectorWriter + from tht.adapters.factory import build_vector_loader - return RestVectorWriter(VectorRestClient(cfg.vector_write_rest), table=table) - - from tht.db.connection import make_engine - from tht.vectorstore.store import VectorStore - - engine = make_engine(cfg.vector_db) - return VectorStore( - engine, schema=cfg.vector_db.db_schema, table=table, dim=cfg.embeddings.dim - ) + return build_vector_loader(cfg, table) def open_searcher(cfg): """Searcher per la LETTURA (similarity search): via REST se `vector_rest` è configurato, altrimenti connessione diretta (dev/test).""" - if cfg.vector_rest is not None: - from tht.vectorstore.reader import RestSearcher - from tht.vectorstore.rest_client import VectorRestClient + from tht.adapters.factory import build_vector_store + from tht.vectorstore.reader import tables_for_kinds - return RestSearcher(VectorRestClient(cfg.vector_rest)) - from tht.db.connection import make_engine - from tht.vectorstore.reader import DirectSearcher + store = build_vector_store(cfg) - return DirectSearcher( - make_engine(cfg.vector_db), schema=cfg.vector_db.db_schema, dim=cfg.embeddings.dim - ) + class AdapterSearcher: + def search(self, query_vec, top_n=10, kinds=None): + return store.search( + tables_for_kinds(kinds), query_vec, limit=top_n, kinds=kinds + ) + + return AdapterSearcher() def _print_stats(stats) -> None: