148 lines
6.4 KiB
Python
148 lines
6.4 KiB
Python
"""Central construction of deployment-specific adapters."""
|
|
|
|
from tht.adapters.dwh import PostgresDwhAdapter, ThothRestDwhAdapter
|
|
from tht.adapters.evidence import FilesystemEvidenceSource, HttpManifestEvidenceSource
|
|
from tht.adapters.evidence.s3 import S3EvidenceSource
|
|
from tht.adapters.vector import PgVectorStore, 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,
|
|
statement_timeout_ms=cfg.execution.statement_timeout_ms,
|
|
)
|
|
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 write capability."""
|
|
resource = cfg.vectors
|
|
if resource is None:
|
|
raise ConfigError("Risorsa vectors non configurata")
|
|
|
|
match resource.type:
|
|
case "pgvector_direct":
|
|
reader = resource.reader or resource.connection
|
|
# Legacy server workspaces use one RW `vector_db` connection. Keep
|
|
# that deployment contract without turning a workstation's legacy
|
|
# compatibility connection into an implicit writer.
|
|
writer = resource.writer or (
|
|
resource.connection if cfg.profile == "server" else None
|
|
)
|
|
if require_write and writer is None:
|
|
raise ConfigError("Vector writer non configurato per pgvector_direct")
|
|
return PgVectorStore(
|
|
reader,
|
|
writer,
|
|
expected_dimension=cfg.embeddings.dim if cfg.embeddings is not None else None,
|
|
)
|
|
case "thoth_vector_http":
|
|
if require_write and resource.writer is None:
|
|
raise ConfigError("Vector writer non configurato")
|
|
return ThothHttpVectorStore(
|
|
VectorRestClient(resource.reader) if resource.reader is not None else None,
|
|
VectorRestClient(resource.writer) if resource.writer is not None else None,
|
|
expected_dimension=cfg.embeddings.dim if cfg.embeddings 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.writer or 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,
|
|
)
|
|
|
|
|
|
def build_evidence_sources(cfg: Config):
|
|
"""Build configured Evidence sources, including the legacy curated filesystem tree."""
|
|
evidence = cfg.evidence
|
|
if evidence is None:
|
|
return []
|
|
sources = []
|
|
if evidence.source_root is not None:
|
|
sources.append(FilesystemEvidenceSource(evidence.source_root / evidence.evidence_dir))
|
|
for resource in evidence.sources:
|
|
match resource.type:
|
|
case "filesystem":
|
|
sources.append(
|
|
FilesystemEvidenceSource(
|
|
resource.root,
|
|
patterns=resource.patterns,
|
|
max_bytes=resource.max_bytes,
|
|
)
|
|
)
|
|
case "http":
|
|
sources.append(
|
|
HttpManifestEvidenceSource(
|
|
[url.get_secret_value() for url in resource.urls],
|
|
connect_timeout=resource.connect_timeout,
|
|
read_timeout=resource.read_timeout,
|
|
max_bytes=resource.max_bytes,
|
|
max_redirects=resource.max_redirects,
|
|
allow_private_hosts=resource.allow_private_hosts,
|
|
max_cache_bytes=resource.max_cache_bytes,
|
|
)
|
|
)
|
|
case "s3":
|
|
def secret(value):
|
|
return value.get_secret_value() if value is not None else None
|
|
|
|
sources.append(S3EvidenceSource(
|
|
bucket=resource.bucket, prefix=resource.prefix,
|
|
endpoint_url=resource.endpoint_url, region=resource.region,
|
|
access_key=secret(resource.access_key), secret_key=secret(resource.secret_key),
|
|
session_token=secret(resource.session_token),
|
|
trusted_endpoint=resource.trusted_endpoint,
|
|
allow_private_endpoint=resource.allow_private_endpoint,
|
|
allow_insecure_endpoint=resource.allow_insecure_endpoint,
|
|
max_bytes=resource.max_bytes, max_objects=resource.max_objects,
|
|
max_pages=resource.max_pages, page_size=resource.page_size,
|
|
))
|
|
case other: # pragma: no cover - Pydantic rejects unsupported discriminators.
|
|
raise ConfigError(f"Adapter evidence non supportato: {other}")
|
|
return sources
|
|
|
|
|
|
__all__ = ["build_dwh", "build_evidence_sources", "build_vector_loader", "build_vector_store"]
|