refactor(core): route integrations through adapter factory

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)

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"]

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":
+    adapter = build_dwh(cfg)
-        _ping_rest(cfg, schema)
+    if isinstance(adapter, ThothRestDwhAdapter):
+        _ping_rest(adapter, cfg, schema)
         return
     engine = make_engine(cfg.database)
     try:

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.adapters.factory import build_dwh
-        from tht.db.sampling import unique_values_for_lsh_rest
+    from tht.db.sampling import SkippedColumn, TruncatedColumn, is_text_type
-        from tht.rest.client import RestClient
+    from tht.mschema.eligibility import effective_eligibility
 
-        values, skipped, truncated = unique_values_for_lsh_rest(
+    dwh = build_dwh(cfg)
-            RestClient(cfg.rest), physical, cfg.lsh, annotations
+    values: dict[str, dict[str, list[str]]] = {}
-        )
+    skipped: list[SkippedColumn] = []
-    else:
+    truncated: list[TruncatedColumn] = []
-        from tht.db.connection import make_engine
+    for table_name, table in physical.tables.items():
-        from tht.db.sampling import unique_values_for_lsh
+        table_ann = annotations.tables.get(table_name)
-
+        for column_name, column in table.columns.items():
-        values, skipped, truncated = unique_values_for_lsh(
+            ann_col = table_ann.columns.get(column_name) if table_ann else None
-            make_engine(cfg.database), physical, cfg.lsh, annotations
+            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:

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)

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.sampling import is_text_type
-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)")
@@ -64,30 +61,23 @@ 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(
+        dwh = build_dwh(cfg)
-                client, database=cfg.database.database, schema=cfg.database.db_schema
+        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,
                         )
-            add_examples_rest(client, phys, cfg.examples)
+                        if value not in (None, "")
+                    ]
         classify_all(phys, cfg.eligibility)
-        except RestError as e:
+    except Exception 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)
     phys.to_yaml(out)

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)."""
+    """EXPLAIN through the configured DWH adapter."""
-    if cfg.database.transport == "rest":
+    from tht.adapters.factory import build_dwh
-        from tht.rest.execute import explain_rest
 
-        return explain_rest(_rest_client(cfg), sql)
+    return build_dwh(cfg).explain(sql)
-    from tht.execute import explain
-
-    return explain(_ro_engine(cfg), sql, timeout_ms=cfg.execution.statement_timeout_ms)
 
 
 def _run_transport(cfg, sql: str, *, limit: int):
-    """Dispatch all'esecutore controllato secondo il transport (direct|rest)."""
+    """Dispatch through the configured DWH adapter."""
-    if cfg.database.transport == "rest":
+    from tht.adapters.factory import build_dwh
-        from tht.rest.execute import run_controlled_rest
 
-        return run_controlled_rest(_rest_client(cfg), sql, limit=limit)
+    return build_dwh(cfg).run_query(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
-    )
 
 
 def do_run(cfg, sql: str, *, limit: int, offset: int = 0):

harness/tht/cli/vector_cmd.py

@@ -46,36 +46,27 @@ 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.adapters.factory import build_vector_loader
-        from tht.vectorstore.rest_client import VectorRestClient
-        from tht.vectorstore.rest_writer import RestVectorWriter
 
-        return RestVectorWriter(VectorRestClient(cfg.vector_write_rest), table=table)
+    return build_vector_loader(cfg, 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
-    )
 
 
 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.adapters.factory import build_vector_store
-        from tht.vectorstore.reader import RestSearcher
+    from tht.vectorstore.reader import tables_for_kinds
-        from tht.vectorstore.rest_client import VectorRestClient
 
-        return RestSearcher(VectorRestClient(cfg.vector_rest))
+    store = build_vector_store(cfg)
-    from tht.db.connection import make_engine
-    from tht.vectorstore.reader import DirectSearcher
 
-    return DirectSearcher(
+    class AdapterSearcher:
-        make_engine(cfg.vector_db), schema=cfg.vector_db.db_schema, dim=cfg.embeddings.dim
+        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:
     typer.secho(