diff --git a/docs/development/plugins.md b/docs/development/plugins.md index b09d4cc..fa4791c 100644 --- a/docs/development/plugins.md +++ b/docs/development/plugins.md @@ -76,12 +76,12 @@ The concept-embedding row and the free-text-embedding row look similar but are n ### Assumptions this table is making -- **Vocabulary only, so far.** Every example above reads OMOP *vocabulary* i.e. operations over concepts, concept relationships, concept embeddings. Nothing about `context.cdm_engine` restricts a plugin to vocabulary tables specifically, however. These tables are available over the same CDM connection groundworkers itself uses, and a plugin *could* query clinical/person-level tables (`condition_occurrence`, `drug_exposure`, etc.) through it just as easily, subject to whatever environmental and procedural access controls actually govern that data. -- **Plugin graph access is TODO.** `PluginContext` does not expose graph traversal at this time, which means that a plugin that wants graph traversal would have to depend on `omop-graph` itself and build its own adapter, reusing `context.cdm_engine` for the connection but not groundworkers' own shared instance. Recommend not doing it that way, however, as it would be easier to expose the core graph the same way `vector_store` and `embedding_backend_factory` already are. +- **Vocabulary only, so far.** Every example above reads OMOP *vocabulary* i.e. operations over concepts, concept relationships, concept embeddings, or the host's shared grounding cascade. Nothing about `context.cdm_engine` restricts a plugin to vocabulary tables specifically, however. These tables are available over the same CDM connection groundworkers itself uses, and a plugin *could* query clinical/person-level tables (`condition_occurrence`, `drug_exposure`, etc.) through it just as easily, subject to whatever environmental and procedural access controls actually govern that data. +- **Shared mapping access.** `PluginContext.mapping_context` is an optional,read-only capability façade over the host's mapping and grounding policy. Use its operations when a plugin needs candidate bundles, text grounding, or concept context. `concept_shared_standard_ancestor(concept_ids, max_depth=5)` returns the unique nearest shared standard ancestor when one exists; raise `max_depth` to widen the hierarchy search. The underlying vocabulary, graph, and service objects are deliberately not exposed. Do not construct a second `omop-graph` or vocabulary adapter from `context.cdm_engine`. The field is `None` only when the host did not compose services for the plugin context. ## Build and register a new plugin -`build()` receives `PluginContext`. The context provides the resolved CDM database and engine, vector store, shared lazy model backend factories, and a deliberately narrow resolver for independent named OA resources. Return `None` when a prerequisite is unavailable. Groundworkers then keeps serving its core capabilities and reports why the plugin was not activated. +`build()` receives `PluginContext`. The context provides the resolved CDM database and engine, vector store, shared lazy model backend factories, the optional read-only `mapping_context`, and a deliberately narrow resolver for independent named OA resources. Return `None` when a prerequisite is unavailable. Groundworkers then keeps serving its core capabilities and reports why the plugin was not activated. `register()` adds MCP tools, prompts, or resources against the state returned by `build()`. The shared MCP registration wrapper supplies the same safe error translation used by core tools. Code-defined prompts can continue to derive their arguments from the callable signature. A plugin publishing data-defined prompts can pass explicit string metadata with `PromptArgument`; this keeps dynamic pack fields out of Python closure signatures: diff --git a/docs/services/vocab.md b/docs/services/vocab.md index 3371f63..ff1b391 100644 --- a/docs/services/vocab.md +++ b/docs/services/vocab.md @@ -48,11 +48,29 @@ vocab.search_normalized( vocabulary_id: str | None = None, standard_only: bool = False, limit: int = 20, + allow_unindexed_scan: bool = False, ) -> list[ConceptMatch] ``` Normalized search: lowercases and strips punctuation from both the query and concept names before matching. Catches common surface-form differences (abbreviations, spacing, case) that exact search misses without the false-positive risk of full-text search. +#### Why this one needs a selective filter + +Comparison is on a *computed* expression, so an ordinary index on `concept_name` cannot serve it. Without a selective filter the planner scans every concept. Measured against a 1.7M-concept CDM: + +| Call | Time | +| --- | --- | +| unconstrained | 40s | +| `domain='Drug'` | 25s | +| `vocabulary_id='MedDRA'` | 0.4s | + +So the call is **refused** unless one of the following holds: + +- `vocabulary_id` or `parent_ids` is given. Note that `domain` does **not** count — it looks like a narrowing filter but the largest domains hold millions of concepts, which is why `domain='Drug'` still took 25s. +- `allow_unindexed_scan=True`, which accepts the scan deliberately. + +It raises rather than warning because the unindexed query does not fail — it succeeds slowly, so a warning gets absorbed into a caller that merely looks sluggish. + ### `search_fulltext` ```python @@ -79,32 +97,32 @@ Synonym FTS is included when the `concept_synonym_name_tsvector` sidecar is also ```python vocab.navigate_to_standard( concept_ids: list[int], -) -> list[StandardMapping] +) -> list[ConceptMappingResult] ``` Batch navigation: given a list of concept IDs (which may be non-standard), returns their standard OMOP equivalents by following `"Maps to"` relationship edges. Two queries total regardless of input list size. -Concepts that are already standard are returned as self-mappings (`relationship_id="self"`). Concepts with no `"Maps to"` edge return an empty `standard_concepts` list. Concept IDs not found in the vocabulary are silently omitted. +Concepts that are already standard are returned as self-mappings (`relationship_id="self"`). Concepts with no `"Maps to"` edge have an empty `targets` list. Concept IDs not found in the vocabulary are silently omitted. ### `navigate_to_value` ```python vocab.navigate_to_value( concept_ids: list[int], -) -> list[RelatedConceptMapping] +) -> list[ConceptMappingResult] ``` -Same batch navigation pattern as `navigate_to_standard`, but follows `"Maps to value"` relationship edges. Used for value-domain mapping in OMOP measurement and observation workflows. +Same batch navigation pattern as `navigate_to_standard`, but follows `"Maps to value"` relationship edges. Used for value-domain mapping in OMOP measurement and observation workflows. Results use the same `ConceptMappingResult` shape, with targets in `targets`. ### `navigate_to_unit` ```python vocab.navigate_to_unit( concept_ids: list[int], -) -> list[RelatedConceptMapping] +) -> list[ConceptMappingResult] ``` -Batch navigation following `"Maps to unit"` relationship edges. +Batch navigation following `"Maps to unit"` relationship edges. Results use the same `ConceptMappingResult` shape, with targets in `targets`. ## Return types @@ -126,15 +144,16 @@ class ConceptMatch: ts_rank: float | None # only set by search_fulltext ``` -### `StandardMapping` +### `ConceptMappingResult` ```python @dataclass -class StandardMapping: +class ConceptMappingResult: source_concept_id: int source_concept_name: str source_standard_concept: bool - standard_concepts: list[MappedConcept] + source_classification_concept: bool + targets: list[MappedConcept] ``` ### `MappedConcept` @@ -142,20 +161,19 @@ class StandardMapping: ```python @dataclass class MappedConcept: - concept_id: int - concept_name: str - vocabulary_id: str - domain_id: str - concept_class_id: str - relationship_id: str # "Maps to", "Maps to value", "Maps to unit", or "self" + concept: dict[str, Any] # concept payload, including identity and quality fields + relationship_id: str # "Maps to", "Maps to value", "Maps to unit", or "self" ``` +The direct Python API uses `targets` for all three navigation methods. MCP serialization names that list `standard_concepts` for `navigate_to_standard` and `related_concepts` for value and unit navigation. + ## Serialization helpers `VocabService` provides `serialise_*` helpers that convert the typed return objects to plain dicts suitable for MCP tool responses: - `serialise_concept_match(match: ConceptMatch) -> dict` -- `serialise_standard_mapping(mapping: StandardMapping) -> dict` +- `serialise_standard_mapping(mapping: ConceptMappingResult) -> dict` +- `serialise_related_concept_mapping(mapping: ConceptMappingResult) -> dict` The search MCP tools in `search_tools.py` use these helpers to convert service results into the wire format. Direct Python consumers can use the dataclasses directly without serializing. diff --git a/pyproject.toml b/pyproject.toml index 72825b1..caf9abc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,7 +56,7 @@ embedding-faiss = [ dev = [ "httpx>=0.27,<1", "ipython>=8.0", - "tornado>=6.5.5", + "tornado>=6.5.9", "pytest>=9.0.3", "pytest-cov>=4.0", "mypy>=1.8", @@ -89,7 +89,12 @@ packages = ["src/groundworkers"] [tool.uv] cache-keys = [{ file = "pyproject.toml" }, { git = { commit = true, tags = true } }] -constraint-dependencies = ["setuptools>=83.0.0"] +constraint-dependencies = [ + "setuptools>=83.0.0", + "httpx2>=2.12.0", + "urllib3>=2.8.0", + "PyJWT>=2.15.0", +] [tool.ruff] src = ["src", "tests"] diff --git a/src/groundworkers/adapters/omop_emb.py b/src/groundworkers/adapters/omop_emb.py index e5b58c5..8ed1ae3 100644 --- a/src/groundworkers/adapters/omop_emb.py +++ b/src/groundworkers/adapters/omop_emb.py @@ -162,8 +162,15 @@ def get_neighbours( concept_id: int, limit: int, model_name: str | None, + standard_only: bool = True, + active_only: bool = True, ) -> dict[str, Any]: - """Return stored-vector neighbours without calling the model provider.""" + """Return stored-vector neighbours without calling the model provider. + + The pgvector HNSW index is partial on ``is_standard AND is_valid``. This + method defaults to that indexed set. Pass ``standard_only=False`` to include + non-standard neighbours, at the cost of a full scan. + """ record = self._resolve_model_record(model_name) reader = self._build_reader(record) @@ -183,7 +190,10 @@ def get_neighbours( vector = np.asarray(vectors[concept_id], dtype=float).reshape(1, -1) raw = reader.get_nearest_concepts( query_embedding=vector, - concept_filter=self._build_concept_filter(), + concept_filter=self._build_concept_filter( + standard_only=standard_only, + active_only=active_only, + ), k=limit + 1, faiss_index_config=record.index_config, ) diff --git a/src/groundworkers/app.py b/src/groundworkers/app.py index e56782c..eb224fd 100644 --- a/src/groundworkers/app.py +++ b/src/groundworkers/app.py @@ -15,6 +15,7 @@ from groundworkers.plugins import ( GroundworkersPlugin, GroundworkersPluginReadiness, + MappingContext, PluginConfigResolver, PluginContext, PluginReadinessResult, @@ -52,8 +53,12 @@ class Adapters: # _build_plugin_context) so a plugin reuses the one backend connection # instead of opening a second one. None when the corresponding backend # is unconfigured, same gating as the adapters above. - embedding_backend_factory: Callable[[], Any] | None = field(default=None, repr=False) - embedding_model_backend_factory: Callable[[], Any] | None = field(default=None, repr=False) + embedding_backend_factory: Callable[[], Any] | None = field( + default=None, repr=False + ) + embedding_model_backend_factory: Callable[[], Any] | None = field( + default=None, repr=False + ) chat_backend_factory: Callable[[], Any] | None = field(default=None, repr=False) @@ -104,7 +109,9 @@ def get_model_backend() -> ModelBackend: nonlocal shared_model_backend if shared_model_backend is None: if resolved_model is None: - raise RuntimeError("No embedding model is configured for Groundworkers.") + raise RuntimeError( + "No embedding model is configured for Groundworkers." + ) from omop_llm import build_model_backend_from_resolved shared_model_backend = build_model_backend_from_resolved(resolved_model) @@ -180,9 +187,14 @@ def get_chat_backend() -> ModelBackend: def build_services(config: AppConfig, adapters: Adapters) -> Services: services = Services() assisted_classifier = None - if adapters.llm is not None and config.groundworkers.source_planning_llm_assisted_enabled: + if ( + adapters.llm is not None + and config.groundworkers.source_planning_llm_assisted_enabled + ): assisted_classifier = AssistedColumnRoleClassifier(adapters.llm) - services.source_planning = SourcePlanningService(assisted_classifier=assisted_classifier) + services.source_planning = SourcePlanningService( + assisted_classifier=assisted_classifier + ) if adapters.omop_graph is not None: # cdm is optional; only the classified-edge traversals # (concept_associations / concept_extended_inheritance) require it. @@ -206,7 +218,27 @@ def build_services(config: AppConfig, adapters: Adapters) -> Services: return services -def _build_plugin_context(config: AppConfig, adapters: Adapters) -> PluginContext: +def build_mapping_context( + config: AppConfig, + *, + adapters: Adapters | None = None, + services: Services | None = None, +) -> MappingContext: + """Build the mapping capabilities shared by hosted and standalone plugins.""" + + shared_adapters = adapters or build_adapters(config) + shared_services = services or build_services(config, shared_adapters) + return MappingContext( + mapping_service=shared_services.mapping, + grounding_service=shared_services.grounding, + ) + + +def _build_plugin_context( + config: AppConfig, + adapters: Adapters, + services: Services | None = None, +) -> PluginContext: """Assemble the resolved handles every plugin is given. `cdm_database`/`vector_store` come straight off `AppConfig`, already @@ -223,6 +255,11 @@ def _build_plugin_context(config: AppConfig, adapters: Adapters) -> PluginContex embedding_backend_factory=adapters.embedding_backend_factory, embedding_model_backend_factory=adapters.embedding_model_backend_factory, chat_backend_factory=adapters.chat_backend_factory, + mapping_context=( + build_mapping_context(config, adapters=adapters, services=services) + if services is not None + else None + ), ) @@ -292,7 +329,8 @@ def verify_plugin_readiness( summary="This plugin does not expose readiness verification.", ) adapters = build_adapters(config) - context = _build_plugin_context(config, adapters) + services = build_services(config, adapters) + context = _build_plugin_context(config, adapters, services) state, issue = _build_plugin(config, context, plugin) if state is None: return PluginReadinessResult( @@ -342,13 +380,14 @@ def load_plugin_readiness( def build_application(config: AppConfig) -> GroundworkersApp: adapters = build_adapters(config) - context = _build_plugin_context(config, adapters) + services = build_services(config, adapters) + context = _build_plugin_context(config, adapters, services) plugin_definitions = tuple(discover_plugins()) plugins, plugin_issues = _build_plugins(config, context, plugin_definitions) return GroundworkersApp( config=config, adapters=adapters, - services=build_services(config, adapters), + services=services, plugins=plugins, plugin_definitions=plugin_definitions, plugin_issues=plugin_issues, diff --git a/src/groundworkers/application/setup/databases.py b/src/groundworkers/application/setup/databases.py index 298a852..39a2798 100644 --- a/src/groundworkers/application/setup/databases.py +++ b/src/groundworkers/application/setup/databases.py @@ -35,7 +35,10 @@ DiagnosticSeverity, ResourceDiagnostic, ) -from groundworkers.base.sql import quote_identifier +from groundworkers.base.sql import ( + normalization_key_expression_sql, + quote_identifier, +) from groundworkers.config import GroundworkersConfig CDM_TABLES = ( @@ -88,6 +91,23 @@ ("idx_concept_synonym_lower_name_trgm",), ), ) +# (table, column, label, index name). The indexed K expression is not restated +# here: it is rendered from base.sql.normalization_key_expression, the same +# builder search_normalized queries with, so it cannot drift. +GROUNDWORKERS_NORMALIZED_INDEX_TARGETS = ( + ( + "concept", + "concept_name", + "concept normalized name", + "idx_concept_name_normkey", + ), + ( + "concept_synonym", + "concept_synonym_name", + "concept_synonym normalized name", + "idx_concept_synonym_name_normkey", + ), +) # Taken from omop-emb rather than restated: these describe tables omop-emb owns # and creates, so a column added there must not need a matching edit here to stay # checkable. @@ -111,6 +131,9 @@ def resolve_database_targets( cdm = resolver.resolve_database(groundworkers.cdm_db) if not isinstance(cdm, ResolvedCDMDatabase): return () + cdm_schema = cdm.schema_name or ( + "main" if cdm.connection.url.startswith("sqlite") else "public" + ) embedding_target: DatabaseTarget | None = None expected_embedding_model_name: str | None = None embedding_safe_url: str | None = None @@ -146,7 +169,7 @@ def resolve_database_targets( database_entry_name=cdm.name, connection_name=cdm.connection.name, safe_url=cdm.connection.safe_url, - cdm_schema=cdm.schema_name or "main", + cdm_schema=cdm_schema, vocabulary_schema=cdm.vocab_schema, connection_url=cdm.connection.url, role="cdm", @@ -157,7 +180,7 @@ def resolve_database_targets( database_entry_name=cdm.name, connection_name=cdm.connection.name, safe_url=cdm.connection.safe_url, - cdm_schema=cdm.schema_name or "main", + cdm_schema=cdm_schema, vocabulary_schema=cdm.vocab_schema, connection_url=cdm.connection.url, role="graph", @@ -168,7 +191,7 @@ def resolve_database_targets( database_entry_name=cdm.name, connection_name=cdm.connection.name, safe_url=cdm.connection.safe_url, - cdm_schema=cdm.schema_name or "main", + cdm_schema=cdm_schema, vocabulary_schema=cdm.vocab_schema, connection_url=cdm.connection.url, role="groundworkers", @@ -437,6 +460,11 @@ def _groundworkers_diagnostics( ) ) + if connection.dialect.name == "postgresql": + diagnostics.extend( + _normalized_index_diagnostics(connection, schema, existing_tables) + ) + diagnostics.extend( _groundworkers_embedding_model_diagnostics( target, @@ -478,9 +506,7 @@ def _groundworkers_embedding_model_diagnostics( "Groundworkers grounding model could not be checked because the omop-emb model registry table is missing.", ), ) - rows = _embedding_registry_rows( - connection, schema=target.embedding_schema - ) + rows = _embedding_registry_rows(connection, schema=target.embedding_schema) except Exception as exc: # Broad except: reported as a redacted warning. failure = classify_connection_error(exc) @@ -779,6 +805,83 @@ def _functional_lower_index_exists( return any(index.get("name") in aliases for index in indexes) +def _normalized_index_diagnostics( + connection: Connection, + schema: str | None, + existing_tables: set[str], +) -> tuple[ResourceDiagnostic, ...]: + targets = tuple( + target + for target in GROUNDWORKERS_NORMALIZED_INDEX_TARGETS + if target[0] in existing_tables + ) + if not targets: + return () + statuses = tuple( + (label, normalized_index_status(connection, schema, table, column)) + for table, column, label, _name in targets + ) + unchecked = tuple(label for label, present in statuses if present is None) + if unchecked: + return ( + _info( + "normalized_indexes_unchecked", + "Normalized-name indexes unchecked because PostgreSQL index listings " + f"could not be read: {', '.join(unchecked)}.", + ), + ) + missing = tuple(label for label, present in statuses if not present) + if missing: + return ( + _warning( + "normalized_indexes_missing", + "Optional normalized-name indexes are missing or out of date: " + f"{', '.join(missing)}. Groundworkers can still run, but unbounded " + "normalized searches scan the whole vocabulary table.", + ), + ) + return ( + _info( + "normalized_indexes_present", + "Optional normalized-name indexes are present: " + f"{', '.join(label for _table, _column, label, _name in targets)}.", + ), + ) + + +def normalized_index_matches( + connection: Connection, + schema: str | None, + table_name: str, + column_name: str, +) -> bool: + """True when an index on *table_name* covers the shared K expression.""" + return normalized_index_status(connection, schema, table_name, column_name) is True + + +def normalized_index_status( + connection: Connection, + schema: str | None, + table_name: str, + column_name: str, +) -> bool | None: + """Return True/False for the K index, or None when listing failed.""" + definitions = _postgres_index_definitions(connection, schema, table_name) + if definitions is None: + return None + expected = _squash_index_sql( + normalization_key_expression_sql(connection, column_name) + ) + return any(expected in _squash_index_sql(indexdef) for indexdef in definitions) + + +def _squash_index_sql(sql: str) -> str: + # pg_get_indexdef adds ::text casts and parentheses the compiled expression + # lacks; drop both, plus whitespace and identifier quotes, on both sides. + squashed = sql.lower().replace("::text", "") + return "".join(ch for ch in squashed if ch not in ' \t\n"()') + + def _trigram_lower_index_exists( connection: Connection, inspector: Any, @@ -811,6 +914,21 @@ def _postgres_lower_index_exists( *, required_fragment: str | None = None, ) -> bool: + return any( + _index_defines_lower_expression( + indexdef, + column_name, + required_fragment=required_fragment, + ) + for indexdef in (_postgres_index_definitions(connection, schema, table_name) or ()) + ) + + +def _postgres_index_definitions( + connection: Connection, + schema: str | None, + table_name: str, +) -> tuple[str, ...] | None: try: rows = connection.execute( text( @@ -824,16 +942,9 @@ def _postgres_lower_index_exists( {"schema": schema or "public", "table_name": table_name}, ) except Exception: - # Broad except: fall back to reflected index names. - return False - return any( - _index_defines_lower_expression( - str(row[0]), - column_name, - required_fragment=required_fragment, - ) - for row in rows - ) + # Broad except: callers fall back to reflected names or report absence. + return None + return tuple(str(row[0]) for row in rows) def _index_defines_lower_expression( @@ -903,10 +1014,7 @@ def _column_has_null_values( column = quote_identifier(connection, column_name) return bool( connection.execute( - text( - f"SELECT EXISTS (SELECT 1 FROM {qualified} " - f"WHERE {column} IS NULL)" - ) + text(f"SELECT EXISTS (SELECT 1 FROM {qualified} WHERE {column} IS NULL)") ).scalar() ) diff --git a/src/groundworkers/application/setup/performance_maintenance.py b/src/groundworkers/application/setup/performance_maintenance.py index f118957..8e483f6 100644 --- a/src/groundworkers/application/setup/performance_maintenance.py +++ b/src/groundworkers/application/setup/performance_maintenance.py @@ -5,15 +5,23 @@ import argparse import shutil import sys -from collections.abc import Sequence +from collections.abc import Iterator, Sequence +from contextlib import contextmanager from enum import StrEnum from pathlib import Path +import sqlalchemy as sa from groundskeeping.contracts import Command, CommandPlan, CommandStep from oa_configurator import Resolver +from omop_alchemy.cdm.model.vocabulary import Concept, Concept_Synonym from sqlalchemy import text +from sqlalchemy.engine import Connection from groundworkers.application.setup.configuration import load_configuration +from groundworkers.application.setup.databases import ( + GROUNDWORKERS_NORMALIZED_INDEX_TARGETS, + normalized_index_matches, +) from groundworkers.application.setup.maintenance_runs import ( MaintenanceRun, MaintenanceRunner, @@ -21,12 +29,22 @@ from groundworkers.application.setup.models import ( ConfigurationSnapshot, ) -from groundworkers.base.sql import quote_identifier +from groundworkers.base.sql import ( + normalization_key_expression, + quote_identifier, +) from groundworkers.config import GroundworkersConfig +_MODULE_NAME = ( + __spec__.name + if __spec__ is not None + else "groundworkers.application.setup.performance_maintenance" +) + class PerformanceRemediation(StrEnum): TRIGRAM_INDEXES = "trigram_indexes" + NORMALIZED_INDEXES = "normalized_indexes" EMBEDDING_INDEX = "embedding_index" @@ -38,24 +56,29 @@ def build_performance_commands( ) -> tuple[Command, ...]: """Build safe, out-of-process commands for the selected indexes.""" - environment = () if config_path is None else (("OA_CONFIG_PATH", str(Path(config_path).expanduser())),) + environment = ( + () + if config_path is None + else (("OA_CONFIG_PATH", str(Path(config_path).expanduser())),) + ) commands: list[Command] = [] selected = set(actions) - if PerformanceRemediation.TRIGRAM_INDEXES in selected: - commands.append( - Command( - argv=( - sys.executable, - "-m", - "groundworkers.application.setup.performance_maintenance", - "trigram", - ), - environment=environment, + for remediation, action in ( + (PerformanceRemediation.TRIGRAM_INDEXES, "trigram"), + (PerformanceRemediation.NORMALIZED_INDEXES, "normalized"), + ): + if remediation in selected: + commands.append( + Command( + argv=(sys.executable, "-m", _MODULE_NAME, action), + environment=environment, + ) ) - ) if PerformanceRemediation.EMBEDDING_INDEX in selected: if not embedding_model: - raise ValueError("A registered embedding model is required to build its index.") + raise ValueError( + "A registered embedding model is required to build its index." + ) commands.append( Command( argv=( @@ -97,8 +120,11 @@ def start_performance_run( ) -def populate_trigram_indexes(snapshot: ConfigurationSnapshot) -> None: - """Create the optional PostgreSQL trigram indexes Groundworkers can use.""" +@contextmanager +def _postgres_cdm( + snapshot: ConfigurationSnapshot, *, purpose: str +) -> Iterator[tuple[Connection, str | None, str]]: + """Yield a transaction on the PostgreSQL CDM, its schema, and a table prefix.""" if snapshot.stack is None: raise ValueError("A usable stack configuration is required.") @@ -107,35 +133,96 @@ def populate_trigram_indexes(snapshot: ConfigurationSnapshot) -> None: engine = database.create_engine() try: if engine.dialect.name != "postgresql": - raise ValueError("Trigram indexes require a PostgreSQL CDM database.") + raise ValueError(f"{purpose} require a PostgreSQL CDM database.") schema = getattr(database, "schema_name", None) prefix = f"{quote_identifier(engine, schema)}." if schema else "" with engine.begin() as connection: - connection.execute(text("CREATE EXTENSION IF NOT EXISTS pg_trgm")) - for table, column, index_name in ( - ("concept", "concept_name", "idx_concept_lower_name_trgm"), - ("concept_synonym", "concept_synonym_name", "idx_concept_synonym_lower_name_trgm"), - ): - connection.execute( - text( - f"CREATE INDEX IF NOT EXISTS {quote_identifier(connection, index_name)} " - f"ON {prefix}{quote_identifier(connection, table)} USING gin " - f"(lower({quote_identifier(connection, column)}) gin_trgm_ops)" - ) - ) + yield connection, schema, prefix finally: engine.dispose() +def populate_trigram_indexes(snapshot: ConfigurationSnapshot) -> None: + """Create the optional PostgreSQL trigram indexes Groundworkers can use.""" + + with _postgres_cdm(snapshot, purpose="Trigram indexes") as (connection, _, prefix): + connection.execute(text("CREATE EXTENSION IF NOT EXISTS pg_trgm")) + for table, column, index_name in ( + ("concept", "concept_name", "idx_concept_lower_name_trgm"), + ( + "concept_synonym", + "concept_synonym_name", + "idx_concept_synonym_lower_name_trgm", + ), + ): + connection.execute( + text( + f"CREATE INDEX IF NOT EXISTS {quote_identifier(connection, index_name)} " + f"ON {prefix}{quote_identifier(connection, table)} USING gin " + f"(lower({quote_identifier(connection, column)}) gin_trgm_ops)" + ) + ) + + +def populate_normalized_indexes(snapshot: ConfigurationSnapshot) -> None: + """Create the expression indexes that serve ``search_normalized``. + + The indexed expression is rendered from the query's own builder, so the + index matches on any PostgreSQL setup regardless of plan caching. An index + of the same name built from an older expression is replaced, not kept. + """ + + model_tables = {"concept": Concept, "concept_synonym": Concept_Synonym} + with _postgres_cdm(snapshot, purpose="Normalized-name indexes") as ( + connection, + schema, + prefix, + ): + for table, column, _label, index_name in GROUNDWORKERS_NORMALIZED_INDEX_TARGETS: + model_table = model_tables[table] + light_table = sa.Table( + table, + sa.MetaData(), + sa.Column(column, sa.String), + schema=model_table.__table__.schema, + ) + old_name = ( + "idx_concept_name_normalized" + if table == "concept" + else "idx_concept_synonym_name_normalized" + ) + connection.execute( + sa.schema.DropIndex( + sa.Index(old_name, light_table.c[column]), + if_exists=True, + ) + ) + if normalized_index_matches(connection, schema, table, column): + continue + index = sa.Index( + index_name, + normalization_key_expression(light_table.c[column]), + ) + connection.execute( + sa.schema.DropIndex(index, if_exists=True) + ) + connection.execute(sa.schema.CreateIndex(index)) + qualified = f"{prefix}{quote_identifier(connection, table)}" + connection.exec_driver_sql(f"ANALYZE {qualified}") + + def _main(argv: Sequence[str] | None = None) -> int: parser = argparse.ArgumentParser() - parser.add_argument("action", choices=("trigram",)) + parser.add_argument("action", choices=("trigram", "normalized")) parser.add_argument("--config", type=Path) args = parser.parse_args(argv) snapshot = load_configuration(config_path=args.config) if not snapshot.usable: raise SystemExit("The selected stack configuration is not usable.") - populate_trigram_indexes(snapshot) + if args.action == "normalized": + populate_normalized_indexes(snapshot) + else: + populate_trigram_indexes(snapshot) return 0 @@ -146,6 +233,7 @@ def _main(argv: Sequence[str] | None = None) -> int: __all__ = [ "PerformanceRemediation", "build_performance_commands", + "populate_normalized_indexes", "populate_trigram_indexes", "start_performance_run", ] diff --git a/src/groundworkers/application/setup/runtime_setup.py b/src/groundworkers/application/setup/runtime_setup.py index 758ad1a..bd71e17 100644 --- a/src/groundworkers/application/setup/runtime_setup.py +++ b/src/groundworkers/application/setup/runtime_setup.py @@ -498,4 +498,3 @@ def _optional_str(value: object) -> str | None: def _effective_tool(stack: StackConfig, name: str): return stack.tools.get(name) - diff --git a/src/groundworkers/base/errors.py b/src/groundworkers/base/errors.py index 8ff6fe3..a21bf37 100644 --- a/src/groundworkers/base/errors.py +++ b/src/groundworkers/base/errors.py @@ -10,6 +10,7 @@ "INVALID_INPUT", "BACKEND_UNAVAIL", "QUERY_ERROR", + "QUERY_TIMEOUT", "FORMAT_BINARY_DECODE", "FORMAT_UNRECOGNISED", "MISSING_DEPENDENCY", diff --git a/src/groundworkers/base/sql.py b/src/groundworkers/base/sql.py index 8d6461a..9b0e019 100644 --- a/src/groundworkers/base/sql.py +++ b/src/groundworkers/base/sql.py @@ -2,9 +2,48 @@ from __future__ import annotations +from enum import StrEnum + +from sqlalchemy import column as sql_column +from sqlalchemy import func, literal from sqlalchemy.engine import Connection, Engine -__all__ = ["effective_schema", "quote_identifier"] +__all__ = [ + "KEY_DROP_TOKENS", + "KEY_DROP_TOKEN_PATTERN", + "STOP_PHRASE_SQL_PATTERN", + "NormalizationProfile", + "effective_schema", + "normalization_expression", + "normalization_index_expression_sql", + "normalization_key_expression", + "normalization_key_expression_sql", + "normalized_key", + "quote_identifier", +] + + +class NormalizationProfile(StrEnum): + """Supported normalization contracts shared by Python and SQL callers.""" + + VERBATIM = "verbatim" + AGGRESSIVE = "aggressive" + DRUG_NAME = "drug_name" + + +#: Stop phrases stripped before comparison, as a PostgreSQL regex. +STOP_PHRASE_SQL_PATTERN = r"\m(?:nos|nec|nfs|unspecified|unknown|other|w/?o|w/)\M" + +# Lossless coarse-key invariant: every normalization variant may only delete +# tokens from this set relative to lowercase + non-alphanumeric-to-space. +# Extending a variant beyond this set requires extending the K expression too. +KEY_DROP_TOKENS = frozenset( + { + "nos", "nec", "nfs", "unspecified", "unknown", "other", "w", "o", + "wo", "extended", "modified", "release", + } +) +KEY_DROP_TOKEN_PATTERN = "(?:" + "|".join(sorted(KEY_DROP_TOKENS)) + ")" def quote_identifier(bind: Engine | Connection, name: str) -> str: @@ -28,3 +67,99 @@ def effective_schema(bind: Engine | Connection) -> str | None: comes from, so reflection and generated SQL cannot disagree. """ return bind.get_execution_options().get("schema_translate_map", {}).get(None) + + +def normalization_expression( + column, + *, + profile: str | NormalizationProfile, + remove_stop_phrases: bool, +): + """Build the SQL expression used for normalized comparison. + + The regex patterns and flags are rendered inline rather than bound: a + prepared statement's generic plan cannot match an expression index whose + constants arrive as parameters, so bound patterns silently force a full + scan once the driver starts reusing statements. Search applies this exact + comparison together with the shared lossless coarse-key expression. + """ + profile = NormalizationProfile(profile) + space = _inline(" ") + globally = _inline("g") + expr = func.lower(column) + if remove_stop_phrases: + expr = func.regexp_replace( + expr, _inline(STOP_PHRASE_SQL_PATTERN), space, globally + ) + expr = func.regexp_replace(expr, _inline(r"[^a-z0-9]+"), space, globally) + if profile == "drug_name": + expr = func.replace(expr, _inline(" extended release "), space) + expr = func.replace(expr, _inline(" modified release "), space) + expr = func.regexp_replace(expr, _inline(r"\s+"), space, globally) + return func.btrim(expr) + + +def normalization_index_expression_sql( + bind: Engine | Connection, + column_name: str, + *, + profile: str | NormalizationProfile = NormalizationProfile.VERBATIM, + remove_stop_phrases: bool = True, +) -> str: + """Render :func:`normalization_expression` as index DDL for *bind*'s dialect. + + Compile against a connected dialect: literal escaping follows the server's + ``standard_conforming_strings``, and a never-connected dialect doubles the + regex backslashes, producing an index no query can match. The default + arguments are ``search_normalized``'s defaults, the variant callers use. + """ + expression = normalization_expression( + sql_column(column_name), + profile=profile, + remove_stop_phrases=remove_stop_phrases, + ) + return str( + expression.compile( + dialect=bind.dialect, + compile_kwargs={"literal_binds": True}, + ) + ) + + +def normalized_key(normalised_query: str) -> str: + """Build the lossless coarse key for an already-normalized query.""" + return " ".join( + token for token in normalised_query.split() if token not in KEY_DROP_TOKENS + ) + + +def normalization_key_expression_sql( + bind: Engine | Connection, column_name: str +) -> str: + """Render K(column), the shared coarse-key expression used by all variants.""" + expression = normalization_key_expression(sql_column(column_name)) + return str( + expression.compile( + dialect=bind.dialect, + compile_kwargs={"literal_binds": True}, + ) + ) + + +def normalization_key_expression(column): + """Build K(column), which removes only tokens normalization may delete.""" + space = _inline(" ") + globally = _inline("g") + base = func.regexp_replace( + func.lower(column), _inline(r"[^a-z0-9]+"), space, globally + ) + padded = _inline(" ") + func.btrim(base) + _inline(" ") + token_pattern = f" {KEY_DROP_TOKEN_PATTERN}(?= )" + dropped = func.regexp_replace(padded, _inline(token_pattern), _inline(""), globally) + collapsed = func.regexp_replace(dropped, _inline(r"\s+"), space, globally) + expression = func.btrim(collapsed) + return expression + + +def _inline(value: str): + return literal(value, literal_execute=True) diff --git a/src/groundworkers/plugins.py b/src/groundworkers/plugins.py index 425b394..cfdfec0 100644 --- a/src/groundworkers/plugins.py +++ b/src/groundworkers/plugins.py @@ -37,6 +37,9 @@ MutationOperation, ) + from groundworkers.services.grounding import ConceptGroundingService + from groundworkers.services.mapping import MappingService + logger = logging.getLogger(__name__) _PLUGIN_SETUP_FIELD_KINDS = frozenset( @@ -167,6 +170,79 @@ def resolve_vector_store(self, name: str) -> ResolvedVectorStore: return self._resolver.resolve_vector_store(name) +class MappingContext: + """Stable, read-only access to Groundworkers' shared mapping services. + + The services are deliberately held behind this host-owned façade. Plugins + can reuse the configured vocabulary, graph, and mapping orchestration + without constructing a second adapter or depending on Groundworkers' + application container. All exposed operations are query-only. + """ + + resolver_identity = "groundworkers-grounding-v1" + + def __init__( + self, + *, + mapping_service: MappingService | None = None, + grounding_service: ConceptGroundingService | None = None, + ) -> None: + self._mapping_service = mapping_service + self._grounding_service = grounding_service + + @property + def available(self) -> bool: + """Whether the host built the candidate-mapping service.""" + + return self._mapping_service is not None + + @property + def grounding_available(self) -> bool: + """Whether the shared tiered grounding capability is configured.""" + + return self._grounding_service is not None + + def _require_mapping_service(self) -> MappingService: + if self._mapping_service is None: + raise RuntimeError("Groundworkers mapping service is not configured") + return self._mapping_service + + def candidate_bundle(self, query: str, **kwargs: Any) -> dict[str, Any]: + """Build a shared candidate/evidence bundle for *query*.""" + + return self._require_mapping_service().concept_candidate_bundle(query, **kwargs) + + async def async_candidate_bundle(self, query: str, **kwargs: Any) -> dict[str, Any]: + """Build a candidate bundle using the service's async retrieval path.""" + + return await self._require_mapping_service().async_concept_candidate_bundle( + query, **kwargs + ) + + def concept_mapping_context(self, concept_id: int, **kwargs: Any) -> dict[str, Any]: + """Build the shared context packet for one known OMOP concept.""" + + return self._require_mapping_service().concept_mapping_context( + concept_id, **kwargs + ) + + def concept_shared_standard_ancestor( + self, concept_ids: list[int], *, max_depth: int = 5 + ) -> dict[str, Any]: + """Find a unique nearest common standard ancestor.""" + + return self._require_mapping_service().concept_shared_standard_ancestor( + concept_ids, max_depth=max_depth + ) + + def ground(self, query: str, **kwargs: Any) -> dict[str, Any]: + """Resolve text through Groundworkers' shared tiered grounding policy.""" + + if self._grounding_service is None: + raise RuntimeError("Groundworkers grounding service is not configured") + return self._grounding_service.ground(query, **kwargs) + + @dataclass(frozen=True) class PluginContext: """Resolved, read-only handles passed to every plugin. @@ -190,6 +266,7 @@ class PluginContext: embedding_backend_factory: Callable[[], Any] | None embedding_model_backend_factory: Callable[[], Any] | None chat_backend_factory: Callable[[], Any] | None + mapping_context: MappingContext | None = None class GroundworkersPlugin(Protocol): diff --git a/src/groundworkers/services/mapping.py b/src/groundworkers/services/mapping.py index d91c819..5549318 100644 --- a/src/groundworkers/services/mapping.py +++ b/src/groundworkers/services/mapping.py @@ -5,6 +5,7 @@ from groundworkers.adapters.omop_emb import OmopEmbAdapter from groundworkers.base.errors import GroundworkersError +from groundworkers.base.sql import NormalizationProfile from groundworkers.services.graph import GraphService from groundworkers.services.grounding import ConceptGroundingService from groundworkers.services.vocab import ( @@ -21,6 +22,88 @@ _SYNC_GROUNDING = object() +def _shared_ancestor_failure(concept_ids: list[int], reason: str) -> dict[str, Any]: + return { + "found": False, + "concept_ids": concept_ids, + "selected_ancestor": None, + "selection_reason": reason, + } + + +def _standard_ancestors_for( + graph: GraphService, concept_id: int, max_depth: int +) -> tuple[str | None, dict[int, dict[str, Any]]]: + concept = graph.get_concept(concept_id) + if concept is None: + raise GroundworkersError("NOT_FOUND", f"Concept {concept_id} was not found") + + domain = concept.get("domain_id") + ancestors = ( + {concept_id: {**concept, "depth": 0}} + if concept.get("standard_concept") + else {} + ) + ancestors.update( + { + int(row["concept_id"]): row + for row in graph.get_ancestors(concept_id, max_depth) + if row.get("standard_concept") and row.get("domain_id") == domain + } + ) + return domain, ancestors + + +def _shared_standard_ancestor( + graph: GraphService | None, concept_ids: list[int], max_depth: int +) -> dict[str, Any]: + """Return the unique nearest same-domain standard ancestor, if any. + + Concepts themselves count at depth zero. When equally near ancestors tie, + leave selection unresolved rather than depending on graph row ordering. + """ + + if graph is None: + raise GroundworkersError( + "BACKEND_UNAVAIL", "omop_graph backend is not configured" + ) + ids = list(dict.fromkeys(concept_ids)) + if len(ids) < 2: + return _shared_ancestor_failure(ids, "fewer_than_two_concepts") + + ancestry = [ + _standard_ancestors_for(graph, concept_id, max_depth) for concept_id in ids + ] + domains = {domain for domain, _ in ancestry} + if None in domains or len(domains) != 1: + return _shared_ancestor_failure(ids, "incompatible_domains") + + maps = [ancestors for _, ancestors in ancestry] + shared_ids = set.intersection(*(set(ancestors) for ancestors in maps)) + if not shared_ids: + return _shared_ancestor_failure(ids, "no_shared_standard_ancestor") + + ranked = sorted( + ( + max(ancestors[ancestor_id]["depth"] for ancestors in maps), + ancestor_id, + ) + for ancestor_id in shared_ids + ) + nearest_depth, nearest_id = ranked[0] + if len(ranked) > 1 and ranked[1][0] == nearest_depth: + return _shared_ancestor_failure(ids, "multiple_nearest_shared_ancestors") + + selected = dict(maps[0][nearest_id]) + selected["depth"] = nearest_depth + return { + "found": True, + "concept_ids": ids, + "selected_ancestor": selected, + "selection_reason": "nearest_shared_standard_ancestor", + } + + class MappingService: """Direct Python API for mapping-oriented vocabulary workflows.""" @@ -43,32 +126,41 @@ def concept_search_normalized( *, domain: str | None = None, vocabulary_id: str | None = None, + parent_ids: list[int] | None = None, standard_only: bool = False, include_synonyms: bool = False, - normalization_profile: str = "verbatim", + normalization_profile: str + | NormalizationProfile = NormalizationProfile.VERBATIM, remove_stop_phrases: bool = True, limit: int = 20, + allow_unindexed_scan: bool = False, ) -> dict[str, Any]: stripped = query.strip() if not stripped: raise ValueError("query must be a non-empty string") + if parent_ids is not None and any(parent_id <= 0 for parent_id in parent_ids): + raise ValueError("all parent_ids must be positive integers") normalized_query, normalization_steps = normalize_text_for_matching( query, profile=normalization_profile, remove_stop_phrases=remove_stop_phrases, ) if not normalized_query: - raise ValueError("query must contain searchable content after normalization") + raise ValueError( + "query must contain searchable content after normalization" + ) - results = self._vocab.search_normalized( + results, plan = self._vocab.search_normalized_with_plan( normalized_query, domain=domain or None, vocabulary_id=vocabulary_id or None, + parent_ids=parent_ids, standard_only=standard_only, include_synonyms=include_synonyms, normalization_profile=normalization_profile, remove_stop_phrases=remove_stop_phrases, limit=limit, + allow_unindexed_scan=allow_unindexed_scan, ) serialised = [] for match in results: @@ -82,7 +174,9 @@ def concept_search_normalized( row["matched_text"] = matched_text row["matched_text_normalized"] = matched_text_normalized row["match_mode"] = ( - "synonym_exact_normalized" if match.match_source == "synonym" else "label_exact_normalized" + "synonym_exact_normalized" + if match.match_source == "synonym" + else "label_exact_normalized" ) serialised.append(row) return { @@ -90,9 +184,55 @@ def concept_search_normalized( "normalized_query": normalized_query, "normalization_profile": normalization_profile, "normalization_steps": normalization_steps, + "plan": plan, "results": serialised, } + def _normalized_channel( + self, + query: str, + *, + domain: str | None, + vocabulary_id: str | None, + standard_only: bool, + active_only: bool, + parent_ids: list[int] | None, + limit: int, + ) -> tuple[dict[str, Any] | None, str | None]: + """Run normalized search using the same index/scope policy as direct calls.""" + try: + results, plan = self._vocab.search_normalized_with_plan( + query, + domain=domain or None, + vocabulary_id=vocabulary_id or None, + standard_only=standard_only, + active_only=active_only, + include_synonyms=False, + parent_ids=parent_ids, + limit=limit, + ) + except GroundworkersError as exc: + note = str(exc) + if exc.code not in {"INVALID_INPUT", "QUERY_TIMEOUT"}: + raise + if exc.code == "INVALID_INPUT": + return None, f"normalized channel omitted: {note}" + return ( + {"available": False, "results": [], "retrieval_notes": [note]}, + f"normalized channel timed out: {note}", + ) + return ( + { + "available": True, + "plan": plan, + "results": [serialise_concept_match(r) for r in results], + "retrieval_notes": [ + "deterministic normalized equality over concept labels" + ], + }, + None, + ) + def concept_candidate_bundle( self, query: str, @@ -112,7 +252,9 @@ def concept_candidate_bundle( per_channel_limit: int = 10, overall_limit: int = 30, model_name: str | None = None, - _embedding_result: dict[str, Any] | GroundworkersError | object = _SYNC_EMBEDDING, + _embedding_result: dict[str, Any] + | GroundworkersError + | object = _SYNC_EMBEDDING, ) -> dict[str, Any]: stripped = query.strip() if not stripped: @@ -144,25 +286,25 @@ def concept_candidate_bundle( channels["exact"] = { "available": True, "results": [serialise_concept_match(r) for r in exact_results], - "retrieval_notes": ["case-insensitive exact match over concept_name and optional synonyms"], + "retrieval_notes": [ + "case-insensitive exact match over concept_name and optional synonyms" + ], } if include_normalized: - normalized_results = self._vocab.search_normalized( + channel, warning = self._normalized_channel( query, - domain=domain or None, - vocabulary_id=vocabulary_id or None, + domain=domain, + vocabulary_id=vocabulary_id, standard_only=standard_only, active_only=active_only, - include_synonyms=False, parent_ids=parent_ids, limit=per_channel_limit, ) - channels["normalized"] = { - "available": True, - "results": [serialise_concept_match(r) for r in normalized_results], - "retrieval_notes": ["deterministic normalized equality over concept labels"], - } + if channel is not None: + channels["normalized"] = channel + if warning: + warnings.append(warning) if include_fulltext: fts_results, fts_available = self._vocab.search_fulltext( @@ -178,15 +320,25 @@ def concept_candidate_bundle( channels["fulltext"] = { "available": fts_available, "results": [serialise_concept_match(r) for r in fts_results], - "retrieval_notes": ["ranked PostgreSQL full-text retrieval"] if fts_available else [], + "retrieval_notes": ["ranked PostgreSQL full-text retrieval"] + if fts_available + else [], } if not fts_available: - warnings.append("full-text sidecar columns unavailable; fulltext channel omitted") + warnings.append( + "full-text sidecar columns unavailable; fulltext channel omitted" + ) if include_embedding: if self._emb is None: - channels["embedding"] = {"available": False, "results": [], "retrieval_notes": []} - warnings.append("embedding adapter not configured; embedding channel omitted") + channels["embedding"] = { + "available": False, + "results": [], + "retrieval_notes": [], + } + warnings.append( + "embedding adapter not configured; embedding channel omitted" + ) else: try: if _embedding_result is _SYNC_EMBEDDING: @@ -206,14 +358,20 @@ def concept_candidate_bundle( embedding_result = _embedding_result emb_notes = ["semantic retrieval from omop-emb"] if parent_ids: - emb_notes.append("parent_ids hierarchy filter was not applied at the embedding level") + emb_notes.append( + "parent_ids hierarchy filter was not applied at the embedding level" + ) channels["embedding"] = { "available": True, "results": embedding_result.get("results", []), "retrieval_notes": emb_notes, } except GroundworkersError as exc: - channels["embedding"] = {"available": False, "results": [], "retrieval_notes": []} + channels["embedding"] = { + "available": False, + "results": [], + "retrieval_notes": [], + } warnings.append(f"embedding channel unavailable: {exc.message}") candidate_union = self._build_candidate_union(channels, overall_limit) @@ -221,18 +379,31 @@ def concept_candidate_bundle( standardized_candidates: list[dict[str, Any]] = [] if include_standard_mappings and candidate_union: - concept_ids = [row["concept_id"] for row in candidate_union if not row.get("standard_concept")] + concept_ids = [ + row["concept_id"] + for row in candidate_union + if not row.get("standard_concept") + ] if concept_ids: mappings = self._vocab.navigate_to_standard(concept_ids) - standardized_candidates = [serialise_standard_mapping(m) for m in mappings] - mapping_index = {m["source_concept_id"]: m["standard_concepts"] for m in standardized_candidates} + standardized_candidates = [ + serialise_standard_mapping(m) for m in mappings + ] + mapping_index = { + m["source_concept_id"]: m["standard_concepts"] + for m in standardized_candidates + } for row in candidate_union: - row["mapped_standard_concepts"] = mapping_index.get(row["concept_id"], []) + row["mapped_standard_concepts"] = mapping_index.get( + row["concept_id"], [] + ) if include_hierarchy_context and self._graph is not None: for row in candidate_union[: min(5, len(candidate_union))]: try: - row["ancestor_preview"] = self._graph.get_ancestors(row["concept_id"], 2)[:3] + row["ancestor_preview"] = self._graph.get_ancestors( + row["concept_id"], 2 + )[:3] except Exception: row["ancestor_preview"] = [] elif include_hierarchy_context: @@ -332,13 +503,17 @@ def concept_nearest_standard_ancestor( _grounded_result: dict[str, Any] | object = _SYNC_GROUNDING, ) -> dict[str, Any]: if self._graph is None: - raise GroundworkersError("BACKEND_UNAVAIL", "omop_graph backend is not configured") + raise GroundworkersError( + "BACKEND_UNAVAIL", "omop_graph backend is not configured" + ) if (query is None) == (concept_id is None): raise ValueError("exactly one of query or concept_id must be provided") if query is not None: if self._grounding is None: - raise GroundworkersError("BACKEND_UNAVAIL", "grounding service is not configured") + raise GroundworkersError( + "BACKEND_UNAVAIL", "grounding service is not configured" + ) if _grounded_result is _SYNC_GROUNDING: grounded = self._grounding.ground( query.strip(), @@ -364,8 +539,12 @@ def concept_nearest_standard_ancestor( seed = results[0] seed_concept = self._graph.get_concept(seed["concept_id"]) if seed_concept is None: - raise GroundworkersError("NOT_FOUND", f"Concept {seed['concept_id']} was not found") - seed_is_standard = seed_concept is not None and seed_concept.get("standard_concept") + raise GroundworkersError( + "NOT_FOUND", f"Concept {seed['concept_id']} was not found" + ) + seed_is_standard = seed_concept is not None and seed_concept.get( + "standard_concept" + ) selection_reason = ( "exact_standard_match" if seed.get("match_kind") == "EXACT" and seed_is_standard @@ -375,8 +554,14 @@ def concept_nearest_standard_ancestor( assert concept_id is not None seed_concept = self._graph.get_concept(concept_id) if seed_concept is None: - raise GroundworkersError("NOT_FOUND", f"Concept {concept_id} was not found") - seed = {"concept_id": concept_id, "concept_name": seed_concept["concept_name"], "match_kind": "DIRECT"} + raise GroundworkersError( + "NOT_FOUND", f"Concept {concept_id} was not found" + ) + seed = { + "concept_id": concept_id, + "concept_name": seed_concept["concept_name"], + "match_kind": "DIRECT", + } selection_reason = "direct_concept_input" warnings: list[str] = [] @@ -387,7 +572,7 @@ def concept_nearest_standard_ancestor( ancestors = self._graph.get_ancestors(seed_concept["concept_id"], max_depth) standard_ancestors = [a for a in ancestors if a.get("standard_concept")] selected_parent = standard_ancestors[0] if standard_ancestors else None - alternatives = standard_ancestors[1: min(5, len(standard_ancestors))] + alternatives = standard_ancestors[1 : min(5, len(standard_ancestors))] if selected_parent is None: # No standard ancestor. Before giving up, follow "Maps to": a @@ -409,7 +594,10 @@ def concept_nearest_standard_ancestor( ) path_payload = [] - if selected_parent and selected_parent["concept_id"] != seed_concept["concept_id"]: + if ( + selected_parent + and selected_parent["concept_id"] != seed_concept["concept_id"] + ): path_info = self._graph.find_path( seed_concept["concept_id"], selected_parent["concept_id"], @@ -430,6 +618,12 @@ def concept_nearest_standard_ancestor( "warnings": warnings, } + def concept_shared_standard_ancestor( + self, concept_ids: list[int], *, max_depth: int = 5 + ) -> dict[str, Any]: + """Return a unique nearest common standard ancestor, when one exists.""" + return _shared_standard_ancestor(self._graph, concept_ids, max_depth) + async def async_concept_nearest_standard_ancestor( self, *, @@ -484,7 +678,9 @@ def concept_mapping_context( model_name: str | None = None, ) -> dict[str, Any]: if self._graph is None: - raise GroundworkersError("BACKEND_UNAVAIL", "omop_graph backend is not configured") + raise GroundworkersError( + "BACKEND_UNAVAIL", "omop_graph backend is not configured" + ) if concept_id <= 0: raise ValueError("concept_id must be a positive integer") concept = self._graph.get_concept(concept_id) @@ -498,11 +694,17 @@ def concept_mapping_context( concept["concept_code"], ) if include_ancestors: - result["ancestors"] = self._graph.get_ancestors(concept_id, max(1, min(ancestor_limit, 10)))[:ancestor_limit] + result["ancestors"] = self._graph.get_ancestors( + concept_id, max(1, min(ancestor_limit, 10)) + )[:ancestor_limit] if include_descendants: - result["descendants"] = self._graph.get_descendants(concept_id, max(1, min(descendant_limit, 10)))[:descendant_limit] + result["descendants"] = self._graph.get_descendants( + concept_id, max(1, min(descendant_limit, 10)) + )[:descendant_limit] if include_relationship_summary: - result["relationship_summary"] = self._summarise_edges(self._graph.get_edges(concept_id)) + result["relationship_summary"] = self._summarise_edges( + self._graph.get_edges(concept_id) + ) if include_neighbors: neighbors = self._graph.get_neighbors( concept_id=concept_id, @@ -515,7 +717,9 @@ def concept_mapping_context( if include_embedding_neighbors: if self._emb is None: result["embedding_neighbors"] = [] - result.setdefault("warnings", []).append("embedding adapter not configured") + result.setdefault("warnings", []).append( + "embedding adapter not configured" + ) else: result["embedding_neighbors"] = self._emb.get_neighbours( concept_id=concept_id, @@ -530,20 +734,28 @@ def concept_map_to_value( concept_code: str, ) -> dict[str, Any]: if self._graph is None: - raise GroundworkersError("BACKEND_UNAVAIL", "omop_graph backend is not configured") + raise GroundworkersError( + "BACKEND_UNAVAIL", "omop_graph backend is not configured" + ) if not vocabulary_id.strip(): raise ValueError("vocabulary_id must be a non-empty string") if not concept_code.strip(): raise ValueError("concept_code must be a non-empty string") source_list = self._graph.get_concept_by_code(vocabulary_id, concept_code) if not source_list: - raise GroundworkersError("NOT_FOUND", f"Concept {vocabulary_id}:{concept_code} was not found") + raise GroundworkersError( + "NOT_FOUND", f"Concept {vocabulary_id}:{concept_code} was not found" + ) source = source_list[0] mappings = self._vocab.navigate_to_value([source["concept_id"]]) mapping = mappings[0] if mappings else None return { "source_concept": source, - "maps_to_value": serialise_related_concept_mapping(mapping)["related_concepts"] if mapping else [], + "maps_to_value": serialise_related_concept_mapping(mapping)[ + "related_concepts" + ] + if mapping + else [], } def concept_resolve_mapping_expression( @@ -555,9 +767,17 @@ def concept_resolve_mapping_expression( resolve_to_standard: bool = True, ) -> dict[str, Any]: if self._graph is None: - raise GroundworkersError("BACKEND_UNAVAIL", "omop_graph backend is not configured") + raise GroundworkersError( + "BACKEND_UNAVAIL", "omop_graph backend is not configured" + ) if not items: - return {"expression_items": [], "resolved_concept_ids": [], "resolved_concepts": [], "excluded_concepts": [], "counts": {"resolved": 0, "excluded": 0}} + return { + "expression_items": [], + "resolved_concept_ids": [], + "resolved_concepts": [], + "excluded_concepts": [], + "counts": {"resolved": 0, "excluded": 0}, + } resolved: dict[int, dict[str, Any]] = {} excluded: dict[int, dict[str, Any]] = {} try: @@ -568,7 +788,9 @@ def concept_resolve_mapping_expression( continue concepts_to_apply = [concept] if resolve_to_standard and not concept.get("standard_concept"): - mapped = self._graph.map_to_standard(concept["vocabulary_id"], concept["concept_code"]) + mapped = self._graph.map_to_standard( + concept["vocabulary_id"], concept["concept_code"] + ) mapped_standards = mapped.get("standard_concepts", []) if mapped_standards: concepts_to_apply = mapped_standards @@ -576,10 +798,16 @@ def concept_resolve_mapping_expression( for base in concepts_to_apply: expanded.append(base) if item.get("include_descendants"): - expanded.extend(self._graph.get_descendants(base["concept_id"], 2)) + expanded.extend( + self._graph.get_descendants(base["concept_id"], 2) + ) target = excluded if item.get("exclude") else resolved for entry in expanded: - if domain and entry.get("domain_id") and str(entry["domain_id"]).lower() != domain.lower(): + if ( + domain + and entry.get("domain_id") + and str(entry["domain_id"]).lower() != domain.lower() + ): continue target[int(entry["concept_id"])] = entry if deduplicate: @@ -628,14 +856,22 @@ def mapping_evaluate_candidates( ref = reference_index.pop(key, None) predicted_ids = self._extract_predicted_ids(pred, top_k=top_k) if ref is None: - extra_prediction_cases.append({"source_key": key, "predicted_concept_ids": predicted_ids}) + extra_prediction_cases.append( + {"source_key": key, "predicted_concept_ids": predicted_ids} + ) continue reference_id = int(ref["reference_standard_concept_id"]) - domain_name = str(ref.get("domain_id") or pred.get("domain_id") or "UNKNOWN") + domain_name = str( + ref.get("domain_id") or pred.get("domain_id") or "UNKNOWN" + ) bucket = by_domain_counts.setdefault(domain_name, Counter()) if reference_id in predicted_ids: agreement_cases.append( - {"source_key": key, "reference_standard_concept_id": reference_id, "predicted_concept_ids": predicted_ids} + { + "source_key": key, + "reference_standard_concept_id": reference_id, + "predicted_concept_ids": predicted_ids, + } ) bucket["agreement"] += 1 else: @@ -651,7 +887,9 @@ def mapping_evaluate_candidates( missing_reference_cases = [ { "source_key": key, - "reference_standard_concept_id": int(row["reference_standard_concept_id"]), + "reference_standard_concept_id": int( + row["reference_standard_concept_id"] + ), "domain_id": row.get("domain_id"), } for key, row in reference_index.items() @@ -659,8 +897,12 @@ def mapping_evaluate_candidates( total_compared = len(agreement_cases) + len(disagreement_cases) summary_metrics = { - "accuracy": round(len(agreement_cases) / total_compared, 6) if total_compared else 0.0, - "coverage": round(total_compared / len(reference_mappings), 6) if reference_mappings else 0.0, + "accuracy": round(len(agreement_cases) / total_compared, 6) + if total_compared + else 0.0, + "coverage": round(total_compared / len(reference_mappings), 6) + if reference_mappings + else 0.0, "agreement_count": len(agreement_cases), "disagreement_count": len(disagreement_cases), "missing_reference_count": len(missing_reference_cases), @@ -689,10 +931,11 @@ def mapping_evaluate_candidates( def _backfill_candidate_metadata( self, candidate_union: list[dict[str, Any]], warnings: list[str] ) -> None: - """Fill identity metadata for candidates surfaced only via embedding. - """ + """Fill identity metadata for candidates surfaced only via embedding.""" incomplete_ids = [ - row["concept_id"] for row in candidate_union if row.get("vocabulary_id") is None + row["concept_id"] + for row in candidate_union + if row.get("vocabulary_id") is None ] if not incomplete_ids: return @@ -719,7 +962,9 @@ def _backfill_candidate_metadata( row["standard_concept"] = view.get("standard_concept") @staticmethod - def _build_candidate_union(channels: dict[str, dict[str, Any]], overall_limit: int) -> list[dict[str, Any]]: + def _build_candidate_union( + channels: dict[str, dict[str, Any]], overall_limit: int + ) -> list[dict[str, Any]]: union: dict[int, dict[str, Any]] = {} channel_order = ("exact", "normalized", "fulltext", "embedding") for channel_name in channel_order: @@ -737,7 +982,9 @@ def _build_candidate_union(channels: dict[str, dict[str, Any]], overall_limit: i "vocabulary_id": item.get("vocabulary_id"), "domain_id": item.get("domain_id"), "concept_class_id": item.get("concept_class_id"), - "standard_concept": item.get("standard_concept", item.get("is_standard")), + "standard_concept": item.get( + "standard_concept", item.get("is_standard") + ), "retrieved_by": [], }, ) diff --git a/src/groundworkers/services/vocab.py b/src/groundworkers/services/vocab.py index e1d6322..288507c 100644 --- a/src/groundworkers/services/vocab.py +++ b/src/groundworkers/services/vocab.py @@ -2,6 +2,8 @@ import logging import re +import threading +import time from dataclasses import dataclass, field from typing import Any @@ -20,9 +22,19 @@ from groundworkers.base.concept_payload import serialise_concept_view from groundworkers.base.domain_names import DomainNameResolver from groundworkers.base.errors import GroundworkersError -from groundworkers.base.sql import effective_schema +from groundworkers.base.sql import ( + NormalizationProfile, + effective_schema, + normalization_expression, + normalization_key_expression, + normalized_key, +) logger = logging.getLogger(__name__) +NORMALIZED_SEARCH_TIMEOUT_MS = 15_000 +_NORMALIZED_INDEX_CACHE: dict[tuple[object, str, str], tuple[bool, float]] = {} +_NORMALIZED_INDEX_CACHE_LOCK = threading.Lock() +_NORMALIZED_INDEX_NEGATIVE_TTL = 60.0 # Keep this service focused on vocabulary queries. Reuse an omop-graph primitive # when it already provides the required operation; higher-level policy belongs in @@ -32,16 +44,18 @@ # Return types # --------------------------------------------------------------------------- + @dataclass class ConceptMatch: """A single candidate returned by search_exact, search_normalized, or search_fulltext.""" + #: Shared concept payload from ``base.concept_payload``, so search results #: report the same flags as every other concept-returning tool. Previously a #: parallel field list carrying ``standard_concept`` and raw #: ``invalid_reason`` but no ``classification_concept``, which meant search #: could not distinguish a classification concept from an unflagged one. concept: dict[str, Any] - match_source: str # "name" | "synonym" + match_source: str # "name" | "synonym" matched_synonym: str | None = None ts_rank: float | None = None @@ -93,16 +107,11 @@ class ConceptMappingResult: targets: list[MappedConcept] = field(default_factory=list) -#: Retained so existing type annotations and imports keep working; both -#: navigations now return the same shape. -StandardMapping = ConceptMappingResult -RelatedConceptMapping = ConceptMappingResult - - # --------------------------------------------------------------------------- # Service # --------------------------------------------------------------------------- + class VocabService: """Direct Python API for OMOP vocabulary search and concept navigation. @@ -244,7 +253,9 @@ def search_exact( Concept.domain_id, Concept.concept_class_id, Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), Concept.is_valid_expr().label("is_active"), ).where(func.lower(Concept.concept_name) == q.lower()), domain=domain, @@ -270,12 +281,20 @@ def search_exact( Concept.domain_id, Concept.concept_class_id, Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), Concept.is_valid_expr().label("is_active"), Concept_Synonym.concept_synonym_name, ) - .join(Concept_Synonym, Concept_Synonym.concept_id == Concept.concept_id) - .where(func.lower(Concept_Synonym.concept_synonym_name) == q.lower()), + .join( + Concept_Synonym, + Concept_Synonym.concept_id == Concept.concept_id, + ) + .where( + func.lower(Concept_Synonym.concept_synonym_name) + == q.lower() + ), domain=domain, vocabulary_id=vocabulary_id, standard_only=standard_only, @@ -284,10 +303,16 @@ def search_exact( ).limit(remaining) if seen_ids: - syn_stmt = syn_stmt.where(Concept.concept_id.not_in(list(seen_ids))) + syn_stmt = syn_stmt.where( + Concept.concept_id.not_in(list(seen_ids)) + ) for row in session.execute(syn_stmt).all(): - results.append(_row_to_match(row, "synonym", row.concept_synonym_name, None)) + results.append( + _row_to_match( + row, "synonym", row.concept_synonym_name, None + ) + ) except GroundworkersError: raise @@ -311,15 +336,52 @@ def search_normalized( standard_only: bool = False, active_only: bool = False, include_synonyms: bool = False, - normalization_profile: str = "verbatim", + normalization_profile: str | NormalizationProfile = NormalizationProfile.VERBATIM, parent_ids: list[int] | None = None, remove_stop_phrases: bool = True, limit: int = 20, + allow_unindexed_scan: bool = False, ) -> list[ConceptMatch]: + """Run normalized equality search, preserving the list return contract.""" + results, _plan = self.search_normalized_with_plan( + query, + domain=domain, + vocabulary_id=vocabulary_id, + standard_only=standard_only, + active_only=active_only, + include_synonyms=include_synonyms, + normalization_profile=normalization_profile, + parent_ids=parent_ids, + remove_stop_phrases=remove_stop_phrases, + limit=limit, + allow_unindexed_scan=allow_unindexed_scan, + ) + return results + + def search_normalized_with_plan( + self, + query: str, + *, + domain: str | None = None, + vocabulary_id: str | None = None, + standard_only: bool = False, + active_only: bool = False, + include_synonyms: bool = False, + normalization_profile: str + | NormalizationProfile = NormalizationProfile.VERBATIM, + parent_ids: list[int] | None = None, + remove_stop_phrases: bool = True, + limit: int = 20, + allow_unindexed_scan: bool = False, + ) -> tuple[list[ConceptMatch], str]: """Deterministic near-verbatim search after text normalization. Both the query and candidate text are normalized before comparison. Distinct from full-text search: deterministic equality, not ranked retrieval. + + Filters are optional when the normalized-key index exists. Without it, + pass a domain, vocabulary_id, or parent_ids to allow a bounded-time scan. + Pass ``allow_unindexed_scan=True`` to explicitly allow an unscoped scan. """ normalized_query, _steps = normalize_text_for_matching( query, @@ -329,17 +391,41 @@ def search_normalized( if not normalized_query: raise ValueError("query must be a non-empty string after normalization") + plan = self._normalized_search_plan( + domain=domain, + vocabulary_id=vocabulary_id, + parent_ids=parent_ids, + include_synonyms=include_synonyms, + allow_unindexed_scan=allow_unindexed_scan, + ) + key = normalized_key(normalized_query) + results: list[ConceptMatch] = [] seen_ids: set[int] = set() try: - name_expr = _normalized_sql_expr( + name_expr = normalization_expression( Concept.concept_name, - normalization_profile=normalization_profile, + profile=normalization_profile, remove_stop_phrases=remove_stop_phrases, ) + name_key_expr = normalization_key_expression(Concept.concept_name) with self._cdm.session() as session: + if ( + self._cdm.engine.dialect.name == "postgresql" + and plan != "unindexed_scan" + ): + session.execute( + select( + func.set_config( + "statement_timeout", + str(NORMALIZED_SEARCH_TIMEOUT_MS), + True, + ) + ), + bind_arguments={"mapper": Concept}, + ) name_stmt = self._apply_concept_filters( select( Concept.concept_id, @@ -349,9 +435,11 @@ def search_normalized( Concept.domain_id, Concept.concept_class_id, Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), Concept.is_valid_expr().label("is_active"), - ).where(name_expr == normalized_query), + ).where(name_key_expr == key, name_expr == normalized_query), domain=domain, vocabulary_id=vocabulary_id, standard_only=standard_only, @@ -366,11 +454,14 @@ def search_normalized( if include_synonyms: remaining = limit - len(results) if remaining > 0: - syn_expr = _normalized_sql_expr( + syn_expr = normalization_expression( Concept_Synonym.concept_synonym_name, - normalization_profile=normalization_profile, + profile=normalization_profile, remove_stop_phrases=remove_stop_phrases, ) + syn_key_expr = normalization_key_expression( + Concept_Synonym.concept_synonym_name + ) syn_stmt = self._apply_concept_filters( select( Concept.concept_id, @@ -380,12 +471,17 @@ def search_normalized( Concept.domain_id, Concept.concept_class_id, Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), Concept.is_valid_expr().label("is_active"), Concept_Synonym.concept_synonym_name, ) - .join(Concept_Synonym, Concept_Synonym.concept_id == Concept.concept_id) - .where(syn_expr == normalized_query), + .join( + Concept_Synonym, + Concept_Synonym.concept_id == Concept.concept_id, + ) + .where(syn_key_expr == key, syn_expr == normalized_query), domain=domain, vocabulary_id=vocabulary_id, standard_only=standard_only, @@ -394,19 +490,100 @@ def search_normalized( ).limit(remaining) if seen_ids: - syn_stmt = syn_stmt.where(Concept.concept_id.not_in(list(seen_ids))) + syn_stmt = syn_stmt.where( + Concept.concept_id.not_in(list(seen_ids)) + ) for row in session.execute(syn_stmt).all(): - results.append(_row_to_match(row, "synonym", row.concept_synonym_name, None)) + results.append( + _row_to_match( + row, "synonym", row.concept_synonym_name, None + ) + ) except GroundworkersError: raise except Exception as exc: + if self._is_query_canceled(exc): + raise GroundworkersError( + "QUERY_TIMEOUT", + f"Normalized search exceeded its " + f"{NORMALIZED_SEARCH_TIMEOUT_MS // 1000} second time limit. Add a domain, " + "vocabulary_id, or parent_ids filter, or build indexes via Setup → " + "Performance → normalised-name indexes.", + ) from exc raise GroundworkersError( "QUERY_ERROR", f"search_normalized failed with {type(exc).__name__}." ) from exc - return results + return results, plan + + def _normalized_search_plan( + self, + *, + domain: str | None, + vocabulary_id: str | None, + parent_ids: list[int] | None, + include_synonyms: bool, + allow_unindexed_scan: bool, + ) -> str: + engine = self._cdm.engine + indexed = engine.dialect.name == "postgresql" and self._has_normalized_index( + "concept", "concept_name" + ) + if indexed and include_synonyms: + indexed = self._has_normalized_index( + "concept_synonym", "concept_synonym_name" + ) + if indexed: + return "indexed" + if domain or vocabulary_id or parent_ids: + return "scoped_scan" + if allow_unindexed_scan: + return "unindexed_scan" + raise GroundworkersError( + "INVALID_INPUT", + "Normalized search needs an index or a scope. Build indexes via Setup → " + "Performance → normalised-name indexes, or pass domain, vocabulary_id, " + "or parent_ids.", + ) + + def _has_normalized_index(self, table: str, column_name: str) -> bool: + engine = self._cdm.engine + key = (engine, table, column_name) + now = time.monotonic() + with _NORMALIZED_INDEX_CACHE_LOCK: + cached = _NORMALIZED_INDEX_CACHE.get(key) + if cached and (cached[0] or now - cached[1] < _NORMALIZED_INDEX_NEGATIVE_TTL): + return cached[0] + try: + from groundworkers.application.setup.databases import ( + normalized_index_matches, + ) + + with engine.connect() as connection: + found = normalized_index_matches( + connection, + effective_schema(connection), + table, + column_name, + ) + except Exception: + found = False + with _NORMALIZED_INDEX_CACHE_LOCK: + _NORMALIZED_INDEX_CACHE[key] = (found, now) + return found + + @staticmethod + def _is_query_canceled(exc: Exception) -> bool: + current: Any = exc + while current is not None: + if getattr(current, "sqlstate", None) == "57014" or getattr( + current, "pgcode", None + ) == "57014": + return True + current = getattr(current, "orig", None) or getattr(current, "__cause__", None) + return False # ------------------------------------------------------------------ # search_fulltext @@ -451,25 +628,31 @@ def search_fulltext( name_rank = func.ts_rank(sa_col("concept_name_tsvector"), tsquery) with self._cdm.session() as session: - name_stmt = self._apply_concept_filters( - select( - Concept.concept_id, - Concept.concept_name, - Concept.concept_code, - Concept.vocabulary_id, - Concept.domain_id, - Concept.concept_class_id, - Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), - Concept.is_valid_expr().label("is_active"), - name_rank.label("ts_rank"), - ).where(sa_col("concept_name_tsvector").op("@@")(tsquery)), - domain=domain, - vocabulary_id=vocabulary_id, - standard_only=standard_only, - active_only=active_only, - parent_ids=parent_ids, - ).order_by(name_rank.desc()).limit(limit) + name_stmt = ( + self._apply_concept_filters( + select( + Concept.concept_id, + Concept.concept_name, + Concept.concept_code, + Concept.vocabulary_id, + Concept.domain_id, + Concept.concept_class_id, + Concept.is_standard_expr().label("standard_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), + Concept.is_valid_expr().label("is_active"), + name_rank.label("ts_rank"), + ).where(sa_col("concept_name_tsvector").op("@@")(tsquery)), + domain=domain, + vocabulary_id=vocabulary_id, + standard_only=standard_only, + active_only=active_only, + parent_ids=parent_ids, + ) + .order_by(name_rank.desc()) + .limit(limit) + ) if min_rank > 0.0: name_stmt = name_stmt.where(name_rank >= min_rank) @@ -481,37 +664,63 @@ def search_fulltext( if include_synonyms and self._fts_synonym_sidecar: remaining = limit - len(results) if remaining > 0: - syn_rank = func.ts_rank(sa_col("concept_synonym_name_tsvector"), tsquery) - syn_stmt = self._apply_concept_filters( - select( - Concept.concept_id, - Concept.concept_name, - Concept.concept_code, - Concept.vocabulary_id, - Concept.domain_id, - Concept.concept_class_id, - Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), - Concept.is_valid_expr().label("is_active"), - Concept_Synonym.concept_synonym_name, - syn_rank.label("ts_rank"), + syn_rank = func.ts_rank( + sa_col("concept_synonym_name_tsvector"), tsquery + ) + syn_stmt = ( + self._apply_concept_filters( + select( + Concept.concept_id, + Concept.concept_name, + Concept.concept_code, + Concept.vocabulary_id, + Concept.domain_id, + Concept.concept_class_id, + Concept.is_standard_expr().label( + "standard_concept" + ), + Concept.is_classification_expr().label( + "classification_concept" + ), + Concept.is_valid_expr().label("is_active"), + Concept_Synonym.concept_synonym_name, + syn_rank.label("ts_rank"), + ) + .join( + Concept_Synonym, + Concept_Synonym.concept_id == Concept.concept_id, + ) + .where( + sa_col("concept_synonym_name_tsvector").op("@@")( + tsquery + ) + ), + domain=domain, + vocabulary_id=vocabulary_id, + standard_only=standard_only, + active_only=active_only, + parent_ids=parent_ids, ) - .join(Concept_Synonym, Concept_Synonym.concept_id == Concept.concept_id) - .where(sa_col("concept_synonym_name_tsvector").op("@@")(tsquery)), - domain=domain, - vocabulary_id=vocabulary_id, - standard_only=standard_only, - active_only=active_only, - parent_ids=parent_ids, - ).order_by(syn_rank.desc()).limit(remaining) + .order_by(syn_rank.desc()) + .limit(remaining) + ) if min_rank > 0.0: syn_stmt = syn_stmt.where(syn_rank >= min_rank) if seen_ids: - syn_stmt = syn_stmt.where(Concept.concept_id.not_in(list(seen_ids))) + syn_stmt = syn_stmt.where( + Concept.concept_id.not_in(list(seen_ids)) + ) for row in session.execute(syn_stmt).all(): - results.append(_row_to_match(row, "synonym", row.concept_synonym_name, float(row.ts_rank))) + results.append( + _row_to_match( + row, + "synonym", + row.concept_synonym_name, + float(row.ts_rank), + ) + ) except GroundworkersError: raise @@ -527,11 +736,13 @@ def search_fulltext( # navigate_to_standard # ------------------------------------------------------------------ - def navigate_to_standard(self, concept_ids: list[int]) -> list[StandardMapping]: + def navigate_to_standard( + self, concept_ids: list[int] + ) -> list[ConceptMappingResult]: """Return standard equivalents for a list of concept_ids via "Maps to" relationship edges. - For concept_ids that are already standard: standard_concepts = [self]. - For concept_ids with no outbound "Maps to" relationship: standard_concepts = []. + For concept_ids that are already standard: targets contains a self-mapping. + With no outbound "Maps to" relationship: targets is empty. concept_ids not found in the vocabulary are silently omitted. """ if not concept_ids: @@ -551,7 +762,9 @@ def navigate_to_standard(self, concept_ids: list[int]) -> list[StandardMapping]: Concept.is_valid_expr().label("is_active"), ).where(Concept.concept_id.in_(concept_ids)) - source_rows = {int(r.concept_id): r for r in session.execute(source_stmt).all()} + source_rows = { + int(r.concept_id): r for r in session.execute(source_stmt).all() + } non_standard_ids = [ cid for cid, r in source_rows.items() if not r.standard_concept @@ -570,13 +783,20 @@ def navigate_to_standard(self, concept_ids: list[int]) -> list[StandardMapping]: Concept.domain_id, Concept.concept_class_id, Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), Concept.is_valid_expr().label("is_active"), ) - .join(Concept, Concept.concept_id == Concept_Relationship.concept_id_2) + .join( + Concept, + Concept.concept_id == Concept_Relationship.concept_id_2, + ) .where( Concept_Relationship.concept_id_1.in_(non_standard_ids), - Concept_Relationship.relationship_id.in_(self.IDENTITY_RELATIONSHIP_IDS), + Concept_Relationship.relationship_id.in_( + self.IDENTITY_RELATIONSHIP_IDS + ), Concept_Relationship.is_valid_expr(), Concept.is_standard_expr(), ) @@ -598,7 +818,7 @@ def navigate_to_standard(self, concept_ids: list[int]) -> list[StandardMapping]: f"navigate_to_standard failed with {type(exc).__name__}.", ) from exc - results: list[StandardMapping] = [] + results: list[ConceptMappingResult] = [] for cid in concept_ids: src = source_rows.get(cid) if src is None: @@ -625,11 +845,15 @@ def navigate_to_standard(self, concept_ids: list[int]) -> list[StandardMapping]: return results - def navigate_to_value(self, concept_ids: list[int]) -> list[RelatedConceptMapping]: + def navigate_to_value( + self, concept_ids: list[int] + ) -> list[ConceptMappingResult]: """Return "Maps to value" related concepts for the given concept_ids.""" return self._navigate_relationship(concept_ids, self.VALUE_RELATIONSHIP_IDS) - def navigate_to_unit(self, concept_ids: list[int]) -> list[RelatedConceptMapping]: + def navigate_to_unit( + self, concept_ids: list[int] + ) -> list[ConceptMappingResult]: """Return "Maps to unit" related concepts for the given concept_ids.""" return self._navigate_relationship(concept_ids, self.UNIT_RELATIONSHIP_IDS) @@ -702,7 +926,7 @@ def _navigate_relationship( self, concept_ids: list[int], relationship_ids: frozenset[str], - ) -> list[RelatedConceptMapping]: + ) -> list[ConceptMappingResult]: if not concept_ids: return [] @@ -720,7 +944,9 @@ def _navigate_relationship( Concept.is_valid_expr().label("is_active"), ).where(Concept.concept_id.in_(concept_ids)) - source_rows = {int(r.concept_id): r for r in session.execute(source_stmt).all()} + source_rows = { + int(r.concept_id): r for r in session.execute(source_stmt).all() + } related: dict[int, list[MappedConcept]] = {} if source_rows: @@ -735,12 +961,19 @@ def _navigate_relationship( Concept.domain_id, Concept.concept_class_id, Concept.is_standard_expr().label("standard_concept"), - Concept.is_classification_expr().label("classification_concept"), + Concept.is_classification_expr().label( + "classification_concept" + ), Concept.is_valid_expr().label("is_active"), ) - .join(Concept, Concept.concept_id == Concept_Relationship.concept_id_2) + .join( + Concept, + Concept.concept_id == Concept_Relationship.concept_id_2, + ) .where( - Concept_Relationship.concept_id_1.in_(list(source_rows.keys())), + Concept_Relationship.concept_id_1.in_( + list(source_rows.keys()) + ), Concept_Relationship.relationship_id.in_(relationship_ids), Concept_Relationship.is_valid_expr(), ) @@ -762,7 +995,7 @@ def _navigate_relationship( f"relationship navigation failed with {type(exc).__name__}.", ) from exc - results: list[RelatedConceptMapping] = [] + results: list[ConceptMappingResult] = [] for cid in concept_ids: src = source_rows.get(cid) if src is None: @@ -783,6 +1016,7 @@ def _navigate_relationship( # Serialisation helpers # --------------------------------------------------------------------------- + def serialise_concept_match(match: ConceptMatch) -> dict: """Serialise a ConceptMatch to a JSON-safe dict for MCP tool responses.""" result: dict = { @@ -838,10 +1072,11 @@ def serialise_related_concept_mapping(mapping: ConceptMappingResult) -> dict: def normalize_text_for_matching( text: str, *, - profile: str = "verbatim", + profile: str | NormalizationProfile = NormalizationProfile.VERBATIM, remove_stop_phrases: bool = True, ) -> tuple[str, list[str]]: """Normalize free text into a deterministic matching form.""" + profile = NormalizationProfile(profile) steps: list[str] = ["strip", "lowercase", "collapse_whitespace"] normalised = _WHITESPACE_RE.sub(" ", text.strip().lower()) @@ -862,19 +1097,9 @@ def normalize_text_for_matching( return normalised, steps -def _normalized_sql_expr(column, *, normalization_profile: str, remove_stop_phrases: bool): - expr = func.lower(column) - if remove_stop_phrases: - expr = func.regexp_replace(expr, r'\m(?:nos|nec|nfs|unspecified|unknown|other|w/?o|w/)\M', " ", "g") - expr = func.regexp_replace(expr, r"[^a-z0-9]+", " ", "g") - if normalization_profile == "drug_name": - expr = func.replace(expr, " extended release ", " ") - expr = func.replace(expr, " modified release ", " ") - expr = func.regexp_replace(expr, r"\s+", " ", "g") - return func.btrim(expr) - - -def _row_to_match(row, match_source: str, matched_synonym: str | None, ts_rank: float | None) -> ConceptMatch: +def _row_to_match( + row, match_source: str, matched_synonym: str | None, ts_rank: float | None +) -> ConceptMatch: return ConceptMatch( concept=serialise_concept_view(row, detail="flags"), match_source=match_source, diff --git a/src/groundworkers/tools/embedding_tools.py b/src/groundworkers/tools/embedding_tools.py index d34e768..3251e91 100644 --- a/src/groundworkers/tools/embedding_tools.py +++ b/src/groundworkers/tools/embedding_tools.py @@ -35,8 +35,18 @@ def embedding_index_status() -> dict[str, Any]: return exc.to_dict() @server.tool("embedding_neighbours") - def embedding_neighbours(concept_id: int, limit: int = 10, model_name: str | None = None) -> dict[str, Any]: - """Returns nearest embedding-space neighbours for one OMOP concept.""" + def embedding_neighbours( + concept_id: int, + limit: int = 10, + model_name: str | None = None, + standard_only: bool = True, + active_only: bool = True, + ) -> dict[str, Any]: + """Returns nearest embedding-space neighbours for one OMOP concept. + + The default uses the partial vector index on is_standard AND is_valid. Pass + standard_only=False to include non-standard neighbours, at the cost of a full scan. + """ if concept_id <= 0: return {"error": True, "code": "INVALID_INPUT", "message": "concept_id must be a positive integer"} safe_limit = max(1, min(limit, 50)) @@ -47,6 +57,8 @@ def embedding_neighbours(concept_id: int, limit: int = 10, model_name: str | Non concept_id=concept_id, limit=safe_limit, model_name=model_name, + standard_only=standard_only, + active_only=active_only, ) except GroundworkersError as exc: return exc.to_dict() diff --git a/src/groundworkers/tools/mapping_tools.py b/src/groundworkers/tools/mapping_tools.py index 565f036..b0b0adf 100644 --- a/src/groundworkers/tools/mapping_tools.py +++ b/src/groundworkers/tools/mapping_tools.py @@ -4,26 +4,36 @@ from groundworkers.base.errors import GroundworkersError from groundworkers.base.server import GroundworkersMCPServer +from groundworkers.base.sql import NormalizationProfile from groundworkers.services import MappingService -def register_mapping_tools(server: GroundworkersMCPServer, mapping_service: MappingService) -> None: +def register_mapping_tools( + server: GroundworkersMCPServer, mapping_service: MappingService +) -> None: @server.tool("concept_search_normalized") def concept_search_normalized( query: str, domain: str | None = None, vocabulary_id: str | None = None, + parent_ids: list[int] | None = None, standard_only: bool = False, include_synonyms: bool = False, - normalization_profile: str = "verbatim", + normalization_profile: NormalizationProfile = NormalizationProfile.VERBATIM, remove_stop_phrases: bool = True, limit: int = 20, + allow_unindexed_scan: bool = False, ) -> dict[str, Any]: """Deterministic near-verbatim concept search after text normalization. Both the query and candidate concept names are normalized before comparison, making this more robust to punctuation and whitespace variation than concept_search_exact while remaining fully deterministic (no ranking). + + Filters are optional when the normalized-name index exists. Without it, + pass domain, vocabulary_id, or parent_ids for a time-limited scoped scan. + A domain is useful when known. Set allow_unindexed_scan=True to accept an + unscoped scan deliberately. """ safe_limit = max(1, min(limit, 50)) try: @@ -31,11 +41,13 @@ def concept_search_normalized( query, domain=domain, vocabulary_id=vocabulary_id, + parent_ids=parent_ids, standard_only=standard_only, include_synonyms=include_synonyms, normalization_profile=normalization_profile, remove_stop_phrases=remove_stop_phrases, limit=safe_limit, + allow_unindexed_scan=allow_unindexed_scan, ) except ValueError as exc: return {"error": True, "code": "INVALID_INPUT", "message": str(exc)} diff --git a/src/groundworkers/tui/pages/setup.py b/src/groundworkers/tui/pages/setup.py index df75090..8bd323a 100644 --- a/src/groundworkers/tui/pages/setup.py +++ b/src/groundworkers/tui/pages/setup.py @@ -406,7 +406,9 @@ def row_selected(self, row_key: str, context: PageContext) -> None: row = next((item for item in view.rows if item.key == row_key), None) if row is None or row.detail is None: return - detail = key_value_detail(f"{self._selected_section_title()} detail", row.detail) + detail = key_value_detail( + f"{self._selected_section_title()} detail", row.detail + ) if detail is not None: context.surface.show_detail(self.route.key, detail) @@ -425,17 +427,24 @@ def action_selected(self, action_key: str, context: PageContext) -> None: self._start_verify_all(context) return if action_key == "overview.integration": - output = self._integration_output() if self._session.databases_connected else None + output = ( + self._integration_output() + if self._session.databases_connected + else None + ) if output is None: - context.notify("Verify the CDM before showing integration commands.", severity="warning") + context.notify( + "Verify the CDM before showing integration commands.", + severity="warning", + ) else: - context.notify(f"stdio: {output.stdio_command}\nHTTP: {output.http_command}") + context.notify( + f"stdio: {output.stdio_command}\nHTTP: {output.http_command}" + ) return if action_key == "graph.prepare": context.open_wizard( - GraphMaintenanceWizardController( - self._session, self._graph_readiness() - ) + GraphMaintenanceWizardController(self._session, self._graph_readiness()) ) return if action_key == "performance.refresh": @@ -622,7 +631,9 @@ def verify() -> None: self.run_worker(verify, thread=True, exclusive=True) - def _finish_performance_checks(self, results, coverage, context: PageContext) -> None: + def _finish_performance_checks( + self, results, coverage, context: PageContext + ) -> None: self._session.connection_results = tuple(results) self._session.embedding_coverage = coverage self._show_current_state(context) @@ -698,7 +709,11 @@ def _trigram_available(self) -> bool: def _graph_target_result(self, target_key: str): return next( - (result for result in self._session.connection_results if result.target_key == target_key), + ( + result + for result in self._session.connection_results + if result.target_key == target_key + ), None, ) @@ -716,7 +731,9 @@ def _integration_output(self): return build_integration_output(self._session.configuration) def _integration_ready(self) -> bool: - return self._session.databases_connected and self._integration_output() is not None + return ( + self._session.databases_connected and self._integration_output() is not None + ) def _start_llm_provider_check(self, context: PageContext) -> None: context.notify("Checking LLM provider endpoint and model.") @@ -855,8 +872,7 @@ def _show_section_detail(self, context: PageContext) -> None: ) tail = collapse_tqdm_tail(tail) step_label = ( - f"{step_index + 1}/{run.total} · " - f"{run.steps[step_index].spec.key}" + f"{step_index + 1}/{run.total} · {run.steps[step_index].spec.key}" if step_index is not None else "No step has started" ) @@ -911,7 +927,9 @@ def _copy_selected_run_log(self, context: PageContext) -> None: step_index = _selected_log_step(run) path = None if step_index is None else run.steps[step_index].log_path if path is None: - context.notify("No log output is available for this run.", severity="warning") + context.notify( + "No log output is available for this run.", severity="warning" + ) return try: text = read_log_tail(path) diff --git a/src/groundworkers/tui/presenters/database.py b/src/groundworkers/tui/presenters/database.py index b02ab47..9a2bc21 100644 --- a/src/groundworkers/tui/presenters/database.py +++ b/src/groundworkers/tui/presenters/database.py @@ -99,10 +99,7 @@ def landing( ), ) - result_by_key = { - item.target_key: _database_result(item) - for item in results - } + result_by_key = {item.target_key: _database_result(item) for item in results} rows = [ _target_row(target, result_by_key.get(target.key)) for target in targets ] @@ -154,6 +151,8 @@ def configuration_detail(self, snapshot: ConfigurationSnapshot) -> KeyValueView: "trigram_indexes_missing", "trigram_indexes_present", "trigram_indexes_unchecked", + "normalized_indexes_missing", + "normalized_indexes_present", } ) @@ -171,7 +170,9 @@ def _database_result(result: ConnectionResult) -> ConnectionResult: ) -def _database_results(results: Sequence[ConnectionResult]) -> tuple[ConnectionResult, ...]: +def _database_results( + results: Sequence[ConnectionResult], +) -> tuple[ConnectionResult, ...]: return tuple(_database_result(result) for result in results) diff --git a/src/groundworkers/tui/presenters/performance.py b/src/groundworkers/tui/presenters/performance.py index 1840161..f6dacfd 100644 --- a/src/groundworkers/tui/presenters/performance.py +++ b/src/groundworkers/tui/presenters/performance.py @@ -18,6 +18,45 @@ ) from groundworkers.tui.presenters.base import SetupPresenterBase +_INDEX_DIAGNOSTICS = ( + ( + "performance.graph.fulltext", + "Graph", + "Full-text indexes", + "database.graph", + ( + "fulltext_sidecar_missing", + "fulltext_indexes_missing", + "fulltext_sidecar_unpopulated", + ), + "fulltext_indexes_present", + ), + ( + "performance.graph.functional", + "Graph", + "Functional text indexes", + "database.graph", + ("functional_indexes_missing",), + "functional_indexes_present", + ), + ( + "performance.groundworkers.trigram", + "Groundworkers", + "Trigram indexes", + "database.groundworkers", + ("trigram_indexes_missing", "trigram_indexes_unchecked"), + "trigram_indexes_present", + ), + ( + "performance.groundworkers.normalized", + "Groundworkers", + "Normalized-name indexes", + "database.groundworkers", + ("normalized_indexes_missing",), + "normalized_indexes_present", + ), +) + class PerformancePresenter(SetupPresenterBase): """Present index readiness without mixing it into database setup.""" @@ -86,8 +125,9 @@ def landing( ), ), message=( - "Graph indexes are prepared from Graph Setup. Groundworkers trigram " - "and embedding indexes can be prepared here when their backends support it." + "Graph indexes are prepared from Graph Setup. Groundworkers trigram, " + "normalized-name and embedding indexes can be prepared here when " + "their backends support it." ), ) @@ -99,39 +139,40 @@ def _statuses( embedding_coverage: EmbeddingCoverageReport | None, ) -> tuple[tuple[tuple[str, str, str, str], SemanticStatus], ...]: by_key = {result.target_key: result for result in connections} - graph = by_key.get("database.graph") - groundworkers = by_key.get("database.groundworkers") - graph_fulltext = _diagnostic_outcome( - graph, - ( - "fulltext_sidecar_missing", - "fulltext_indexes_missing", - "fulltext_sidecar_unpopulated", - ), - present_code="fulltext_indexes_present", - ) - graph_functional = _diagnostic_outcome( - graph, - ("functional_indexes_missing",), - present_code="functional_indexes_present", - ) - trigram = _diagnostic_outcome( - groundworkers, - ("trigram_indexes_missing", "trigram_indexes_unchecked"), - present_code="trigram_indexes_present", + index_rows = tuple( + _status_row( + key, + area, + index, + _diagnostic_outcome( + by_key.get(target_key), + missing_codes, + present_code=present_code, + ), + ) + for key, area, index, target_key, missing_codes, present_code in _INDEX_DIAGNOSTICS ) embedding = _embedding_index_outcome( embedding_configuration, embedding_coverage, ) return ( - (("performance.graph.fulltext", "Graph", "Full-text indexes", graph_fulltext[0]), graph_fulltext[1]), - (("performance.graph.functional", "Graph", "Functional text indexes", graph_functional[0]), graph_functional[1]), - (("performance.groundworkers.trigram", "Groundworkers", "Trigram indexes", trigram[0]), trigram[1]), - (("performance.embeddings.index", "Embeddings", "Vector index", embedding[0]), embedding[1]), + *index_rows, + _status_row( + "performance.embeddings.index", "Embeddings", "Vector index", embedding + ), ) +def _status_row( + key: str, + area: str, + index: str, + outcome: tuple[str, SemanticStatus], +) -> tuple[tuple[str, str, str, str], SemanticStatus]: + return ((key, area, index, outcome[0]), outcome[1]) + + def _diagnostic_outcome( result: ConnectionResult | None, missing_codes: tuple[str, ...], @@ -143,7 +184,10 @@ def _diagnostic_outcome( if not result.connected: return "Connection failed", SemanticStatus.ERROR codes = {diagnostic.code: diagnostic for diagnostic in result.diagnostics} - if any(code in codes and codes[code].severity is DiagnosticSeverity.WARNING for code in missing_codes): + if any( + code in codes and codes[code].severity is DiagnosticSeverity.WARNING + for code in missing_codes + ): return "Missing", SemanticStatus.WARNING if present_code in codes: return "Ready", SemanticStatus.OK diff --git a/src/groundworkers/tui/routes.py b/src/groundworkers/tui/routes.py index 8239436..7aaa43b 100644 --- a/src/groundworkers/tui/routes.py +++ b/src/groundworkers/tui/routes.py @@ -6,4 +6,4 @@ purpose="Configure and verify the services Groundworkers uses.", ) -__all__ = ["SETUP_ROUTE"] \ No newline at end of file +__all__ = ["SETUP_ROUTE"] diff --git a/src/groundworkers/tui/wizards/performance_maintenance.py b/src/groundworkers/tui/wizards/performance_maintenance.py index 848f8c7..460eb90 100644 --- a/src/groundworkers/tui/wizards/performance_maintenance.py +++ b/src/groundworkers/tui/wizards/performance_maintenance.py @@ -28,12 +28,22 @@ _FIELD_BY_REMEDIATION = { PerformanceRemediation.TRIGRAM_INDEXES: "create_trigram_indexes", + PerformanceRemediation.NORMALIZED_INDEXES: "create_normalized_indexes", PerformanceRemediation.EMBEDDING_INDEX: "create_embedding_index", } _LABELS = { PerformanceRemediation.TRIGRAM_INDEXES: "Create Groundworkers trigram indexes", + PerformanceRemediation.NORMALIZED_INDEXES: "Create Groundworkers normalized-name indexes", PerformanceRemediation.EMBEDDING_INDEX: "Build the embedding vector index", } +# Both are plain PostgreSQL indexes on the CDM vocabulary tables, so they share +# the setup page's "connected PostgreSQL CDM" availability check. +_POSTGRES_REMEDIATIONS = frozenset( + { + PerformanceRemediation.TRIGRAM_INDEXES, + PerformanceRemediation.NORMALIZED_INDEXES, + } +) class PerformanceMaintenanceWizardController: @@ -65,7 +75,7 @@ def __init__( ) self._selected: set[PerformanceRemediation] = set() if trigram_available: - self._selected.add(PerformanceRemediation.TRIGRAM_INDEXES) + self._selected.update(_POSTGRES_REMEDIATIONS) if self._embedding_available: self._selected.add(PerformanceRemediation.EMBEDDING_INDEX) self._step_index = 0 @@ -103,8 +113,14 @@ def review(self) -> WizardTransition: def apply(self) -> WizardResult: try: commands = build_performance_commands( - tuple(remediation for remediation in PerformanceRemediation if remediation in self._selected), - embedding_model=(self._coverage.index.model_name if self._coverage else None), + tuple( + remediation + for remediation in PerformanceRemediation + if remediation in self._selected + ), + embedding_model=( + self._coverage.index.model_name if self._coverage else None + ), config_path=self._session.configuration.path, ) if not commands: @@ -148,7 +164,7 @@ def _actions_step(self) -> FormStep: kind=FieldKind.BOOLEAN, default=remediation in self._selected, disabled=( - remediation is PerformanceRemediation.TRIGRAM_INDEXES + remediation in _POSTGRES_REMEDIATIONS and not self._trigram_available ) or ( @@ -157,7 +173,7 @@ def _actions_step(self) -> FormStep: ), help=( "Requires a PostgreSQL CDM database." - if remediation is PerformanceRemediation.TRIGRAM_INDEXES + if remediation in _POSTGRES_REMEDIATIONS else "Requires a registered pgvector model." ), ) @@ -171,11 +187,15 @@ def _review_step(self) -> ReviewStep: title="Confirm", review=WizardReview( changes=tuple( - ReviewChange(field=_LABELS[item], before="available", after="will run") + ReviewChange( + field=_LABELS[item], before="available", after="will run" + ) for item in PerformanceRemediation if item in self._selected ), - effects=("Each index build runs as a background command with its own log.",), + effects=( + "Each index build runs as a background command with its own log.", + ), warnings=( "Index builds can be slow and temporarily use substantial database resources.", ), @@ -192,14 +212,18 @@ def _snapshot(self, *, issues: tuple[ValidationIssue, ...] = ()) -> WizardSnapsh step=step, step_index=self._step_index, step_count=len(steps), - values={} if isinstance(step, ReviewStep) else { + values={} + if isinstance(step, ReviewStep) + else { key: remediation in self._selected for remediation, key in _FIELD_BY_REMEDIATION.items() }, issues=issues, can_back=self._step_index > 0, can_next=not isinstance(step, ReviewStep), - can_apply=isinstance(step, ReviewStep) and step.review.ready_to_apply and not issues, + can_apply=isinstance(step, ReviewStep) + and step.review.ready_to_apply + and not issues, expected_revision=self._session.configuration.revision, ) @@ -210,12 +234,21 @@ def _submit_form(self, values: Mapping[str, object]) -> tuple[ValidationIssue, . else: self._selected.discard(remediation) if not self._selected: - return (ValidationIssue("Select at least one index to prepare.", "create_trigram_indexes"),) + return ( + ValidationIssue( + "Select at least one index to prepare.", "create_trigram_indexes" + ), + ) return () def _as_bool(value: object) -> bool: - return (isinstance(value, bool) and value) or str(value).strip().lower() in {"true", "yes", "1", "on"} + return (isinstance(value, bool) and value) or str(value).strip().lower() in { + "true", + "yes", + "1", + "on", + } __all__ = ["PerformanceMaintenanceWizardController"] diff --git a/tests/unit/test_embedding_tools.py b/tests/unit/test_embedding_tools.py index 36cbabf..324e4b3 100644 --- a/tests/unit/test_embedding_tools.py +++ b/tests/unit/test_embedding_tools.py @@ -25,11 +25,24 @@ def index_status(self) -> dict: ], } - def get_neighbours(self, concept_id: int, limit: int, model_name: str | None) -> dict: + def get_neighbours( + self, + concept_id: int, + limit: int, + model_name: str | None, + standard_only: bool = False, + active_only: bool = True, + ) -> dict: self.calls.append( ( "get_neighbours", - {"concept_id": concept_id, "limit": limit, "model_name": model_name}, + { + "concept_id": concept_id, + "limit": limit, + "model_name": model_name, + "standard_only": standard_only, + "active_only": active_only, + }, ) ) return { @@ -114,13 +127,48 @@ def test_embedding_neighbours_calls_adapter_with_clamped_limit(): assert result["query_concept_id"] == 123 assert adapter.calls == [ - ("get_neighbours", {"concept_id": 123, "limit": 50, "model_name": None}) + ( + "get_neighbours", + { + "concept_id": 123, + "limit": 50, + "model_name": None, + "standard_only": True, + "active_only": True, + }, + ) ] +def test_embedding_neighbours_forwards_concept_filters(): + """standard_only/active_only must reach the adapter: the vector index is partial + on is_standard AND is_valid, so dropping them silently costs a full scan.""" + adapter = StubEmbAdapter() + server = build_server(adapter) + + server.call( + "embedding_neighbours", + concept_id=123, + limit=5, + model_name=None, + standard_only=True, + active_only=True, + ) + + assert adapter.calls[0][1]["standard_only"] is True + assert adapter.calls[0][1]["active_only"] is True + + def test_embedding_neighbours_returns_error_dict_for_cava_error(): class NotFoundAdapter(StubEmbAdapter): - def get_neighbours(self, concept_id: int, limit: int, model_name: str | None) -> dict: + def get_neighbours( + self, + concept_id: int, + limit: int, + model_name: str | None, + standard_only: bool = False, + active_only: bool = True, + ) -> dict: raise GroundworkersError("NOT_FOUND", "missing concept") server = build_server(NotFoundAdapter()) diff --git a/tests/unit/test_mapping_context.py b/tests/unit/test_mapping_context.py new file mode 100644 index 0000000..9959987 --- /dev/null +++ b/tests/unit/test_mapping_context.py @@ -0,0 +1,103 @@ +import asyncio + +import pytest + +from groundworkers.app import build_application +from groundworkers.bootstrap import build_app_config_from_stack +from groundworkers.plugins import MappingContext +from tests.support.stack_config import build_cdm_stack + + +class StubMappingService: + def concept_candidate_bundle(self, query: str, **kwargs): + return {"query": query, "kwargs": kwargs} + + async def async_concept_candidate_bundle(self, query: str, **kwargs): + return {"query": query, "kwargs": kwargs, "async": True} + + def concept_mapping_context(self, concept_id: int, **kwargs): + return {"concept_id": concept_id, "kwargs": kwargs} + + def concept_shared_standard_ancestor(self, concept_ids: list[int], **kwargs): + return {"concept_ids": concept_ids, "kwargs": kwargs} + + +class StubGroundingService: + def ground(self, query: str, **kwargs): + return {"query": query, "kwargs": kwargs} + + +def test_mapping_context_delegates_to_the_host_service(): + context = MappingContext(mapping_service=StubMappingService()) + + assert context.available is True + assert context.candidate_bundle("diabetes", limit=3) == { + "query": "diabetes", + "kwargs": {"limit": 3}, + } + assert context.concept_mapping_context(201826, include_ancestors=False) == { + "concept_id": 201826, + "kwargs": {"include_ancestors": False}, + } + assert context.concept_shared_standard_ancestor([1, 2]) == { + "concept_ids": [1, 2], + "kwargs": {"max_depth": 5}, + } + assert context.concept_shared_standard_ancestor([1, 2], max_depth=8) == { + "concept_ids": [1, 2], + "kwargs": {"max_depth": 8}, + } + + +def test_mapping_context_exposes_async_candidate_retrieval(): + context = MappingContext(mapping_service=StubMappingService()) + + assert asyncio.run(context.async_candidate_bundle("diabetes")) == { + "query": "diabetes", + "kwargs": {}, + "async": True, + } + + +def test_mapping_context_reports_unconfigured_service(): + context = MappingContext() + + assert context.available is False + with pytest.raises(RuntimeError, match="mapping service is not configured"): + context.candidate_bundle("diabetes") + with pytest.raises(RuntimeError, match="mapping service is not configured"): + context.concept_shared_standard_ancestor([1, 2]) + + +def test_mapping_context_exposes_grounding_without_exposing_service_internals(): + context = MappingContext(grounding_service=StubGroundingService()) + + assert context.grounding_available is True + assert context.ground("diabetes", limit=3) == { + "query": "diabetes", + "kwargs": {"limit": 3}, + } + assert not hasattr(context, "grounding_service") + + +class ContextPlugin: + name = "context_plugin" + config_cls = None + + def build(self, context, config): + assert config is None + return context.mapping_context + + def register(self, server, state): + del server, state + + +def test_application_wires_one_shared_mapping_context_into_plugins(monkeypatch): + monkeypatch.setattr("groundworkers.app.discover_plugins", lambda: [ContextPlugin()]) + + app = build_application(build_app_config_from_stack(build_cdm_stack())) + + context = app.plugins["context_plugin"] + assert isinstance(context, MappingContext) + assert context.available is True + assert context.grounding_available is True diff --git a/tests/unit/test_mapping_service.py b/tests/unit/test_mapping_service.py index 5b4cebe..c28fd74 100644 --- a/tests/unit/test_mapping_service.py +++ b/tests/unit/test_mapping_service.py @@ -1,7 +1,8 @@ - +from unittest.mock import Mock import pytest +from groundworkers.base.errors import GroundworkersError from groundworkers.services import MappingService from groundworkers.services.vocab import ( ConceptMappingResult, @@ -29,7 +30,7 @@ def search_exact(self, query: str, *, domain=None, vocabulary_id=None, standard_ ) ] - def search_normalized(self, query: str, *, domain=None, vocabulary_id=None, standard_only=False, active_only=False, include_synonyms=False, normalization_profile="verbatim", parent_ids=None, remove_stop_phrases=True, limit=20): + def search_normalized(self, query: str, *, domain=None, vocabulary_id=None, standard_only=False, active_only=False, include_synonyms=False, normalization_profile="verbatim", parent_ids=None, remove_stop_phrases=True, limit=20, allow_unindexed_scan=False): return [ ConceptMatch( concept={ @@ -47,6 +48,21 @@ def search_normalized(self, query: str, *, domain=None, vocabulary_id=None, stan ) ] + def search_normalized_with_plan(self, query: str, **kwargs): + scoped = bool( + kwargs.get("domain") + or kwargs.get("vocabulary_id") + or kwargs.get("parent_ids") + ) + if not scoped and not kwargs.get("allow_unindexed_scan"): + raise GroundworkersError( + "INVALID_INPUT", + "Build indexes via Setup → Performance → normalised-name indexes, or pass domain, vocabulary_id, or parent_ids.", + ) + return self.search_normalized(query, **kwargs), ( + "scoped_scan" if scoped else "unindexed_scan" + ) + def search_fulltext(self, query: str, *, domain=None, vocabulary_id=None, standard_only=False, active_only=False, include_synonyms=True, parent_ids=None, min_rank=0.0, limit=20): return ([ ConceptMatch( @@ -337,18 +353,28 @@ def build_service(graph: StubGraphAdapter | None = None) -> MappingService: def test_concept_search_normalized_returns_normalized_metadata_and_results(): service = build_service() - result = service.concept_search_normalized(" Type-2 Diabetes, NOS ") + result = service.concept_search_normalized( + " Type-2 Diabetes, NOS ", allow_unindexed_scan=True + ) assert result["normalized_query"] == "type 2 diabetes" + assert result["plan"] == "unindexed_scan" assert result["results"][0]["match_mode"] == "label_exact_normalized" assert result["results"][0]["matched_text_normalized"] == "type 2 diabetes mellitus" +def test_concept_search_normalized_accepts_parent_scope(): + result = build_service().concept_search_normalized("diabetes", parent_ids=[101]) + + assert result["plan"] == "scoped_scan" + + def test_concept_candidate_bundle_combines_channels_and_standard_mappings(): service = build_service() result = service.concept_candidate_bundle( "type 2 diabetes", + vocabulary_id="ICD10CM", include_hierarchy_context=True, include_relationship_summary=True, ) @@ -426,6 +452,7 @@ def test_concept_nearest_standard_ancestor_raises_when_no_graph(): service = MappingService(StubVocabAdapter()) from groundworkers.base.errors import GroundworkersError as GWError + try: service.concept_nearest_standard_ancestor(query="diabetes") assert False, "should have raised" @@ -433,6 +460,77 @@ def test_concept_nearest_standard_ancestor_raises_when_no_graph(): assert exc.code == "BACKEND_UNAVAIL" +def test_concept_shared_standard_ancestor_selects_unique_nearest_common_parent(): + graph = Mock() + graph.get_concept.side_effect = lambda concept_id: { + "concept_id": concept_id, + "concept_name": f"Specific finding {concept_id}", + "domain_id": "Condition", + "standard_concept": False, + } + graph.get_ancestors.side_effect = lambda concept_id, _depth: [ + { + "concept_id": 803, + "concept_name": "Shared standard finding", + "concept_code": "803", + "vocabulary_id": "SNOMED", + "domain_id": "Condition", + "concept_class_id": "Clinical Finding", + "standard_concept": True, + "depth": 1, + }, + { + "concept_id": 804, + "concept_name": "More distant shared finding", + "domain_id": "Condition", + "standard_concept": True, + "depth": 2, + }, + ] if concept_id in {801, 802} else [] + + result = build_service(graph).concept_shared_standard_ancestor([801, 802]) + + assert result["found"] is True + assert result["selection_reason"] == "nearest_shared_standard_ancestor" + assert result["selected_ancestor"]["concept_id"] == 803 + assert result["selected_ancestor"]["depth"] == 1 + + +def test_concept_shared_standard_ancestor_leaves_equal_depth_tie_unresolved(): + graph = Mock() + graph.get_concept.side_effect = lambda concept_id: { + "concept_id": concept_id, + "concept_name": f"Concept {concept_id}", + "domain_id": "Condition", + "standard_concept": False, + } + graph.get_ancestors.return_value = [ + { + "concept_id": ancestor_id, + "concept_name": f"Ancestor {ancestor_id}", + "domain_id": "Condition", + "standard_concept": True, + "depth": 1, + } + for ancestor_id in (803, 804) + ] + + result = build_service(graph).concept_shared_standard_ancestor([801, 802]) + + assert result["found"] is False + assert result["selection_reason"] == "multiple_nearest_shared_ancestors" + assert result["selected_ancestor"] is None + + +def test_concept_shared_standard_ancestor_requires_graph(): + service = MappingService(StubVocabAdapter()) + + from groundworkers.base.errors import GroundworkersError as GWError + + with pytest.raises(GWError, match="omop_graph"): + service.concept_shared_standard_ancestor([801, 802]) + + def test_concept_mapping_context_assembles_requested_context(): service = build_service() @@ -496,6 +594,7 @@ def test_mapping_evaluate_candidates_computes_summary_metrics(): # Recording stub for propagation tests (Gap 8 and Gap 1) # --------------------------------------------------------------------------- + class _RecordingVocabAdapter(StubVocabAdapter): """Wraps StubVocabAdapter and records the active_only / parent_ids received per channel.""" @@ -528,10 +627,11 @@ def _service_with_recording_vocab(emb=None): # active_only propagation (Gap 8) # --------------------------------------------------------------------------- + def test_active_only_true_propagated_to_all_lexical_channels(): service, vocab = _service_with_recording_vocab(emb=StubEmbAdapter()) - service.concept_candidate_bundle("diabetes", active_only=True) + service.concept_candidate_bundle("diabetes", vocabulary_id="ICD10CM", active_only=True) assert vocab.received["exact"]["active_only"] is True assert vocab.received["normalized"]["active_only"] is True @@ -541,7 +641,7 @@ def test_active_only_true_propagated_to_all_lexical_channels(): def test_active_only_false_propagated_to_all_lexical_channels(): service, vocab = _service_with_recording_vocab(emb=StubEmbAdapter()) - service.concept_candidate_bundle("diabetes", active_only=False) + service.concept_candidate_bundle("diabetes", vocabulary_id="ICD10CM", active_only=False) assert vocab.received["exact"]["active_only"] is False assert vocab.received["normalized"]["active_only"] is False @@ -552,6 +652,7 @@ def test_active_only_false_propagated_to_all_lexical_channels(): # parent_ids propagation (Gap 1) # --------------------------------------------------------------------------- + def test_parent_ids_propagated_to_all_lexical_channels(): service, vocab = _service_with_recording_vocab(emb=StubEmbAdapter()) @@ -565,7 +666,7 @@ def test_parent_ids_propagated_to_all_lexical_channels(): def test_parent_ids_none_propagated_when_not_specified(): service, vocab = _service_with_recording_vocab(emb=StubEmbAdapter()) - service.concept_candidate_bundle("diabetes") + service.concept_candidate_bundle("diabetes", vocabulary_id="ICD10CM") assert vocab.received["exact"]["parent_ids"] is None assert vocab.received["normalized"]["parent_ids"] is None @@ -625,6 +726,7 @@ def test_embedding_channel_no_parent_ids_note_when_parent_ids_absent(): # Constraints echoed correctly in bundle output # --------------------------------------------------------------------------- + def test_candidate_bundle_constraints_reflect_active_only_and_parent_ids(): service, _ = _service_with_recording_vocab() @@ -724,3 +826,37 @@ def test_nearest_standard_ancestor_still_succeeds_through_ancestry(): assert result["found"] is True assert result["selected_parent"]["concept_id"] == 301 assert result["warnings"] == [] + + +def test_candidate_bundle_omits_the_normalized_channel_when_it_would_scan(): + service = MappingService(StubVocabAdapter(), graph_service=StubGraphAdapter()) + + result = service.concept_candidate_bundle("diabetes") + + assert "normalized" not in result["channels"] + assert any("normalized channel omitted" in warning for warning in result["warnings"]) + + +def test_candidate_bundle_keeps_the_normalized_channel_when_a_vocabulary_bounds_it(): + service = MappingService(StubVocabAdapter(), graph_service=StubGraphAdapter()) + + result = service.concept_candidate_bundle("diabetes", vocabulary_id="SNOMED") + + assert result["channels"]["normalized"]["available"] is True + assert not any("normalized channel omitted" in warning for warning in result["warnings"]) + + +def test_candidate_bundle_keeps_a_failed_timeout_channel_as_unavailable(): + class TimedOutVocab(StubVocabAdapter): + def search_normalized_with_plan(self, _query, **_kwargs): + raise GroundworkersError("QUERY_TIMEOUT", "normalized scan timed out") + + service = MappingService(TimedOutVocab(), graph_service=StubGraphAdapter()) + + result = service.concept_candidate_bundle("diabetes", domain="Condition") + + assert result["channels"]["normalized"]["available"] is False + assert result["channels"]["normalized"]["retrieval_notes"] == [ + "normalized scan timed out" + ] + assert any("normalized channel timed out" in warning for warning in result["warnings"]) diff --git a/tests/unit/test_mapping_tools.py b/tests/unit/test_mapping_tools.py index cb2d906..83b6824 100644 --- a/tests/unit/test_mapping_tools.py +++ b/tests/unit/test_mapping_tools.py @@ -64,16 +64,27 @@ def test_concept_search_normalized_clamps_limit_and_calls_service(): "query": " Diabetes ", "domain": None, "vocabulary_id": None, + "parent_ids": None, "standard_only": False, "include_synonyms": False, "normalization_profile": "verbatim", "remove_stop_phrases": True, "limit": 50, + "allow_unindexed_scan": False, }, ) ] +def test_concept_search_normalized_forwards_parent_scope(): + service = StubMappingService() + server = build_server(service) + + server.call("concept_search_normalized", query="Diabetes", parent_ids=[101]) + + assert service.calls[0][1]["parent_ids"] == [101] + + def test_concept_candidate_bundle_rejects_invalid_input_from_service(): class InvalidService(StubMappingService): async def async_concept_candidate_bundle(self, query: str, **kwargs): diff --git a/tests/unit/test_normalized_index.py b/tests/unit/test_normalized_index.py new file mode 100644 index 0000000..2948e0b --- /dev/null +++ b/tests/unit/test_normalized_index.py @@ -0,0 +1,201 @@ +"""The K expression must preserve every supported normalization variant.""" + +from __future__ import annotations + +import re +from contextlib import contextmanager + +import pytest +import sqlalchemy as sa +from omop_alchemy.cdm.model.vocabulary import Concept, Concept_Synonym +from sqlalchemy.dialects import postgresql + +from groundworkers.application.setup import performance_maintenance +from groundworkers.application.setup.databases import ( + GROUNDWORKERS_NORMALIZED_INDEX_TARGETS, + _normalized_index_diagnostics, + _squash_index_sql, +) +from groundworkers.base.sql import ( + KEY_DROP_TOKENS, + NormalizationProfile, + normalization_expression, + normalization_key_expression, + normalization_key_expression_sql, + normalized_key, +) +from groundworkers.services.vocab import normalize_text_for_matching + + +def _python_key(text: str) -> str: + tokens = re.sub(r"[^a-z0-9]+", " ", text.lower()).split() + return " ".join(token for token in tokens if token not in KEY_DROP_TOKENS) + + +@pytest.mark.parametrize( + ("profile", "remove_stop_phrases"), + [ + (NormalizationProfile.VERBATIM, True), + (NormalizationProfile.VERBATIM, False), + (NormalizationProfile.DRUG_NAME, True), + (NormalizationProfile.DRUG_NAME, False), + ], +) +def test_python_coarse_key_is_lossless_for_all_variants( + profile: str, remove_stop_phrases: bool +) -> None: + names = ( + "Type 2 diabetes mellitus NOS", + "sodium chloride 0.9% w/ sterile water", + "tablet_W/O coating", + "drug extended release", + "drug modified release", + " UNKNOWN ", + "___", + "w/o", + "W/O", + "wo", + "tabs\tand\tspaces", + "é K İ fi", + "extended release", + "modified release", + "", + ) + for name in names: + normalized, _ = normalize_text_for_matching( + name, profile=profile, remove_stop_phrases=remove_stop_phrases + ) + assert normalized_key(normalized) == _python_key(name) + + +def test_key_expression_is_inlined_and_version_tolerant() -> None: + with sa.create_engine("sqlite://").connect() as connection: + expected = normalization_key_expression_sql(connection, "concept_name") + query = str( + normalization_key_expression(sa.column("concept_name")).compile( + dialect=connection.dialect, + compile_kwargs={"literal_binds": True}, + ) + ) + # PostgreSQL versions add casts and parentheses differently in pg_indexes. + indexdef = ( + "CREATE INDEX idx_concept_name_normkey ON public.concept USING btree " + f"(({expected.replace('concept_name', '((concept_name)::text)')}))" + ) + assert _squash_index_sql(expected) == _squash_index_sql(query) + assert _squash_index_sql(expected) in _squash_index_sql(indexdef) + + +def test_normalized_comparison_remains_variant_specific() -> None: + col = sa.column("concept_name") + verbatim = normalization_expression( + col, profile=NormalizationProfile.VERBATIM, remove_stop_phrases=True + ) + drug = normalization_expression( + col, profile=NormalizationProfile.DRUG_NAME, remove_stop_phrases=True + ) + assert "extended release" not in str(drug) + assert "extended release" not in str(verbatim) or str(drug) != str(verbatim) + + +def test_index_names_and_unreadable_listing_diagnostic(monkeypatch) -> None: + names = {target[3] for target in GROUNDWORKERS_NORMALIZED_INDEX_TARGETS} + assert names == { + "idx_concept_name_normkey", + "idx_concept_synonym_name_normkey", + } + + monkeypatch.setattr( + "groundworkers.application.setup.databases.normalized_index_status", + lambda *_args: None, + ) + diagnostics = _normalized_index_diagnostics( + object(), None, {"concept", "concept_synonym"} + ) + + assert diagnostics[0].code == "normalized_indexes_unchecked" + + +def test_normalized_index_ddl_uses_light_tables_and_preserves_orm_metadata( + monkeypatch, +) -> None: + class RecordingConnection: + dialect = postgresql.dialect() + + def __init__(self) -> None: + self.statements = [] + self.raw_statements = [] + + def execute(self, statement): + self.statements.append(statement) + + def exec_driver_sql(self, statement): + self.raw_statements.append(statement) + + connection = RecordingConnection() + + @contextmanager + def fake_postgres_cdm(_snapshot, *, purpose): + assert purpose == "Normalized-name indexes" + yield connection, "cdm", '"cdm".' + + monkeypatch.setattr(performance_maintenance, "_postgres_cdm", fake_postgres_cdm) + monkeypatch.setattr( + performance_maintenance, "normalized_index_matches", lambda *_args: False + ) + concept_indexes_before = set(Concept.__table__.indexes) + synonym_indexes_before = set(Concept_Synonym.__table__.indexes) + + performance_maintenance.populate_normalized_indexes(object()) + + creates = [ + statement + for statement in connection.statements + if isinstance(statement, sa.schema.CreateIndex) + ] + drops = [ + statement + for statement in connection.statements + if isinstance(statement, sa.schema.DropIndex) + ] + assert [statement.element.name for statement in creates] == [ + "idx_concept_name_normkey", + "idx_concept_synonym_name_normkey", + ] + assert {statement.element.name for statement in drops} == { + "idx_concept_name_normalized", + "idx_concept_synonym_name_normalized", + "idx_concept_name_normkey", + "idx_concept_synonym_name_normkey", + } + + compiled_creates = [ + str( + statement.compile( + dialect=postgresql.dialect(), + schema_translate_map={None: "cdm"}, + render_schema_translate=True, + ) + ) + for statement in creates + ] + compiled_drops = [ + str( + statement.compile( + dialect=postgresql.dialect(), + schema_translate_map={None: "cdm"}, + render_schema_translate=True, + ) + ) + for statement in drops + ] + assert "ON cdm.concept (" in compiled_creates[0] + assert "ON cdm.concept_synonym (" in compiled_creates[1] + assert all("regexp_replace" in statement for statement in compiled_creates) + assert all("cdm." in statement for statement in compiled_drops) + assert connection.raw_statements == [ + 'ANALYZE "cdm".concept', + 'ANALYZE "cdm".concept_synonym', + ] + assert Concept.__table__.indexes == concept_indexes_before + assert Concept_Synonym.__table__.indexes == synonym_indexes_before diff --git a/tests/unit/test_performance.py b/tests/unit/test_performance.py index 6b8c5c6..dc12db3 100644 --- a/tests/unit/test_performance.py +++ b/tests/unit/test_performance.py @@ -33,7 +33,11 @@ def test_performance_surface_collects_index_readiness() -> None: "fulltext_indexes_present", "functional_indexes_present", ), - _result("database.groundworkers", "trigram_indexes_present"), + _result( + "database.groundworkers", + "trigram_indexes_present", + "normalized_indexes_present", + ), ), embedding_configuration=None, embedding_coverage=None, @@ -45,21 +49,27 @@ def test_performance_surface_collects_index_readiness() -> None: ("Graph", "Full-text indexes", "Ready"), ("Graph", "Functional text indexes", "Ready"), ("Groundworkers", "Trigram indexes", "Ready"), + ("Groundworkers", "Normalized-name indexes", "Ready"), ("Embeddings", "Vector index", "Not configured"), ] -def test_performance_commands_run_trigram_before_embedding_index() -> None: +def test_performance_commands_run_text_indexes_before_embedding_index() -> None: commands = build_performance_commands( ( - PerformanceRemediation.TRIGRAM_INDEXES, PerformanceRemediation.EMBEDDING_INDEX, + PerformanceRemediation.NORMALIZED_INDEXES, + PerformanceRemediation.TRIGRAM_INDEXES, ), embedding_model="ollama/qwen3-embedding:0.6b", config_path="/tmp/config.toml", ) assert commands[0].argv[-1] == "trigram" - assert commands[1].argv[1:3] == ("maintenance", "rebuild-index") - assert "--index-type" in commands[1].argv - assert all(command.environment == (("OA_CONFIG_PATH", "/tmp/config.toml"),) for command in commands) + assert commands[1].argv[-1] == "normalized" + assert commands[2].argv[1:3] == ("maintenance", "rebuild-index") + assert "--index-type" in commands[2].argv + assert all( + command.environment == (("OA_CONFIG_PATH", "/tmp/config.toml"),) + for command in commands + ) diff --git a/tests/unit/test_performance_maintenance_wizard.py b/tests/unit/test_performance_maintenance_wizard.py new file mode 100644 index 0000000..c542771 --- /dev/null +++ b/tests/unit/test_performance_maintenance_wizard.py @@ -0,0 +1,25 @@ +from pathlib import Path + +from oa_configurator import save_stack_config + +from groundworkers.tui.state import SetupSession +from groundworkers.tui.wizards.performance_maintenance import ( + PerformanceMaintenanceWizardController, +) +from tests.support.stack_config import build_cdm_stack + + +def test_postgres_trigram_action_is_available(tmp_path: Path) -> None: + config_path = tmp_path / "config.toml" + save_stack_config(build_cdm_stack(), config_path) + controller = PerformanceMaintenanceWizardController( + SetupSession(config_path=config_path), + embedding_coverage=None, + trigram_available=True, + ) + + values = controller.start().values + + assert values["create_trigram_indexes"] is True + assert values["create_normalized_indexes"] is True + assert values["create_embedding_index"] is False diff --git a/tests/unit/test_setup_databases.py b/tests/unit/test_setup_databases.py index 4ace03a..71797ee 100644 --- a/tests/unit/test_setup_databases.py +++ b/tests/unit/test_setup_databases.py @@ -83,9 +83,7 @@ def test_distinct_vocabulary_connection_is_refused_rather_than_targeted( snapshot = load_configuration(config_path=path) assert snapshot.state is ConfigurationState.INCOMPLETE - assert [issue.code for issue in snapshot.issues] == [ - "vocabulary_connection_split" - ] + assert [issue.code for issue in snapshot.issues] == ["vocabulary_connection_split"] assert resolve_database_targets(snapshot) == () diff --git a/tests/unit/test_vocab_service.py b/tests/unit/test_vocab_service.py index 7e84d9b..c90cda0 100644 --- a/tests/unit/test_vocab_service.py +++ b/tests/unit/test_vocab_service.py @@ -3,17 +3,21 @@ import re from datetime import date +import pytest from omop_alchemy.cdm.model.vocabulary import ( Concept, Concept_Ancestor, Concept_Relationship, Concept_Synonym, ) +from sqlalchemy import column as sa_column from sqlalchemy import create_engine, event from groundworkers.adapters.cdm import CDMAdapter +from groundworkers.base.errors import GroundworkersError from groundworkers.base.server import GroundworkersMCPServer -from groundworkers.services.vocab import VocabService +from groundworkers.base.sql import normalization_expression +from groundworkers.services.vocab import VocabService, normalize_text_for_matching from groundworkers.tools.search_tools import register_search_tools @@ -48,7 +52,13 @@ def regexp_replace(value, pattern, replacement, _flags): ) connection.execute( Concept_Synonym.__table__.insert(), - ({"concept_id": 1, "concept_synonym_name": "Diabetes", "language_concept_id": 0},), + ( + { + "concept_id": 1, + "concept_synonym_name": "Diabetes", + "language_concept_id": 0, + }, + ), ) connection.execute( Concept_Relationship.__table__.insert(), @@ -88,7 +98,9 @@ def _concept( } -def test_vocab_service_exact_normalized_sidecar_and_standard_navigation(tmp_path) -> None: +def test_vocab_service_exact_normalized_sidecar_and_standard_navigation( + tmp_path, +) -> None: service = _service(tmp_path) exact = service.search_exact("diabetes", include_synonyms=True) @@ -116,7 +128,9 @@ def test_vocab_service_exact_normalized_sidecar_and_standard_navigation(tmp_path assert "concept_code" in target and "is_active" in target -def test_fulltext_probe_uses_the_first_available_concept_label(tmp_path, monkeypatch) -> None: +def test_fulltext_probe_uses_the_first_available_concept_label( + tmp_path, monkeypatch +) -> None: service = _service(tmp_path) queried: list[str] = [] @@ -146,3 +160,169 @@ def test_search_tools_expose_the_characterized_vocab_service(tmp_path) -> None: "results": [], } assert navigation["results"][0]["standard_concepts"][0]["concept_id"] == 1 + + +# --------------------------------------------------------------------------- +# Normalized-search affordability guard +# --------------------------------------------------------------------------- + + +def test_normalized_search_is_refused_when_unscoped_and_unindexed( + tmp_path, +) -> None: + """No index and no scope requires the caller to choose a remedy.""" + service = _service(tmp_path) + + with pytest.raises(GroundworkersError) as excinfo: + service.search_normalized("diabetes mellitus") + + assert "Setup → Performance → normalised-name indexes" in excinfo.value.message + assert "domain, vocabulary_id, or parent_ids" in excinfo.value.message + + +def test_empty_parent_ids_do_not_count_as_a_scope(tmp_path) -> None: + service = _service(tmp_path) + + with pytest.raises(GroundworkersError) as excinfo: + service.search_normalized("diabetes mellitus", parent_ids=[]) + + assert excinfo.value.code == "INVALID_INPUT" + + +def test_a_domain_filter_allows_a_scoped_scan(tmp_path) -> None: + service = _service(tmp_path) + + results = service.search_normalized("diabetes mellitus", domain="Condition") + assert [item.concept_id for item in results] == [1, 2] + + +@pytest.mark.parametrize( + "kwargs", + [ + {"vocabulary_id": "ICD10CM"}, + {"parent_ids": [1]}, + {"domain": "Condition"}, + {"allow_unindexed_scan": True}, + ], + ids=["vocabulary_id", "parent_ids", "domain", "explicit_opt_in"], +) +def test_normalized_search_runs_when_bounded_or_explicitly_allowed( + tmp_path, kwargs +) -> None: + service = _service(tmp_path) + + # Any of these makes the call affordable or accepts the cost deliberately. + assert isinstance(service.search_normalized("diabetes mellitus", **kwargs), list) + + +def test_normalized_search_plan_uses_index_without_filters(tmp_path, monkeypatch) -> None: + service = _service(tmp_path) + monkeypatch.setattr(service._cdm.engine.dialect, "name", "postgresql") + monkeypatch.setattr(service, "_has_normalized_index", lambda *_args: True) + + assert service._normalized_search_plan( + domain=None, + vocabulary_id=None, + parent_ids=None, + include_synonyms=False, + allow_unindexed_scan=False, + ) == "indexed" + + +def test_normalized_search_runs_unfiltered_when_index_is_available( + tmp_path, monkeypatch +) -> None: + service = _service(tmp_path) + monkeypatch.setattr( + service, + "_normalized_search_plan", + lambda **_kwargs: "indexed", + ) + + results, plan = service.search_normalized_with_plan("diabetes mellitus") + + assert plan == "indexed" + assert [item.concept_id for item in results] == [1, 2] + + +def test_normalized_search_translates_postgres_statement_timeout(tmp_path, monkeypatch) -> None: + service = _service(tmp_path) + + class QueryCanceled(RuntimeError): + sqlstate = "57014" + + class FakeSession: + calls = 0 + bind_arguments = None + + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def execute(self, _statement, **kwargs): + self.calls += 1 + if self.calls == 1: + type(self).bind_arguments = kwargs.get("bind_arguments") + return None # SET LOCAL statement_timeout + raise QueryCanceled("canceling statement due to statement timeout") + + monkeypatch.setattr(service._cdm.engine.dialect, "name", "postgresql") + monkeypatch.setattr(service, "_has_normalized_index", lambda *_args: True) + monkeypatch.setattr(service._cdm, "session", FakeSession) + + with pytest.raises(GroundworkersError) as excinfo: + service.search_normalized("diabetes", domain="Condition") + + assert excinfo.value.code == "QUERY_TIMEOUT" + assert "normalised-name indexes" in excinfo.value.message + assert "domain, vocabulary_id, or parent_ids" in excinfo.value.message + assert "15 second time limit" in excinfo.value.message + assert FakeSession.bind_arguments == {"mapper": Concept} + + +def test_explicit_unindexed_scan_skips_postgres_statement_timeout( + tmp_path, monkeypatch +) -> None: + service = _service(tmp_path) + statements = [] + + class FakeResult: + def all(self): + return [] + + class FakeSession: + def __enter__(self): + return self + + def __exit__(self, *_args): + return False + + def execute(self, statement): + statements.append(statement) + return FakeResult() + + monkeypatch.setattr(service._cdm.engine.dialect, "name", "postgresql") + monkeypatch.setattr(service, "_has_normalized_index", lambda *_args: False) + monkeypatch.setattr(service._cdm, "session", FakeSession) + + results, plan = service.search_normalized_with_plan( + "diabetes", allow_unindexed_scan=True + ) + + assert plan == "unindexed_scan" + assert results == [] + assert len(statements) == 1 + assert "SET LOCAL statement_timeout" not in str(statements[0]) + + +def test_unknown_normalization_profile_is_rejected_by_python_and_sql() -> None: + with pytest.raises(ValueError, match="NormalizationProfile"): + normalize_text_for_matching("diabetes", profile="invented") + with pytest.raises(ValueError, match="NormalizationProfile"): + normalization_expression( + sa_column("concept_name"), + profile="invented", + remove_stop_phrases=True, + ) diff --git a/uv.lock b/uv.lock index cbebb4d..dba1d5d 100644 --- a/uv.lock +++ b/uv.lock @@ -11,7 +11,12 @@ resolution-markers = [ ] [manifest] -constraints = [{ name = "setuptools", specifier = ">=83.0.0" }] +constraints = [ + { name = "httpx2", specifier = ">=2.12.0" }, + { name = "pyjwt", specifier = ">=2.15.0" }, + { name = "setuptools", specifier = ">=83.0.0" }, + { name = "urllib3", specifier = ">=2.8.0" }, +] [[package]] name = "airium" @@ -1207,7 +1212,7 @@ requires-dist = [ { name = "sqlalchemy", specifier = ">=2,<3" }, { name = "textual", marker = "extra == 'dev'", specifier = ">=0.60" }, { name = "textual", marker = "extra == 'tui'", specifier = ">=0.60" }, - { name = "tornado", marker = "extra == 'dev'", specifier = ">=6.5.5" }, + { name = "tornado", marker = "extra == 'dev'", specifier = ">=6.5.9" }, { name = "ty", marker = "extra == 'dev'", specifier = ">=0.0.59" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.29,<1.0" }, ] @@ -1289,15 +1294,15 @@ wheels = [ [[package]] name = "httpcore2" -version = "2.11.0" +version = "2.13.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "h11" }, { name = "truststore" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/30/7d/ee6787efd5fe675d7cfd5eb149e40ccb5bdfc7e7c9252edcf7825c38986f/httpcore2-2.11.0.tar.gz", hash = "sha256:82e6fc95d784e6ee22ebd4b2cb57df53a2efb13ad6a11260a236ecebbc5f50c7", size = 67532, upload-time = "2026-08-18T08:03:53.008Z" } +sdist = { url = "https://files.pythonhosted.org/packages/cb/f3/1db7aa2bc2524062192bb0e0323969492d1883152a232fe36eea65f4e35c/httpcore2-2.13.1.tar.gz", hash = "sha256:e0aa977abe17e69a3b820a24542a6fa88702676d83880b8d194dcd18408e5103", size = 68071, upload-time = "2026-09-23T07:47:22.372Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/f3/54/e84a5c82ac0959d5e55b3970d326fd95306446b2bf0702302888745f7e5c/httpcore2-2.11.0-py3-none-any.whl", hash = "sha256:c7c899fbc6b8abb6e747dda427aa6f52934c45e191eabb986965164ebb02a908", size = 83061, upload-time = "2026-08-18T08:03:50.894Z" }, + { url = "https://files.pythonhosted.org/packages/09/ba/a4568248771ce81957bfb7cc600264a40fbcda092391ee1c415c50be4bea/httpcore2-2.13.1-py3-none-any.whl", hash = "sha256:e1e05d4f25f7d7d496bfb96748f6f4b67657b03da069b3a68c36069f3db73d0a", size = 83423, upload-time = "2026-09-23T07:47:19.365Z" }, ] [[package]] @@ -1362,7 +1367,7 @@ wheels = [ [[package]] name = "httpx2" -version = "2.11.0" +version = "2.13.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio", marker = "sys_platform != 'emscripten'" }, @@ -1372,9 +1377,9 @@ dependencies = [ { name = "truststore", marker = "sys_platform != 'emscripten'" }, { name = "typing-extensions", marker = "python_full_version < '3.13'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/d5/4d/b3fcae38f29bfb0f300517d085c488f41f65e5b0a73023976b2122f568cc/httpx2-2.11.0.tar.gz", hash = "sha256:ea01b2e8febfb026e2601814c77ecb1e64fff114a87bc789cb520e67f27e7809", size = 99617, upload-time = "2026-08-18T08:03:53.691Z" } +sdist = { url = "https://files.pythonhosted.org/packages/d5/44/474bef2a0e9d90f1715d32cb98b0738695ca17ba324095fb2497ed7fbd59/httpx2-2.13.1.tar.gz", hash = "sha256:e48744a19e3af5ee48313d0ce5fe941d5422fae5705ea922a4aabf94d7800dfa", size = 100405, upload-time = "2026-09-23T07:47:23.052Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/39/29/f60bcf54028601920c0ce3da81537bcd0af4ee19dd9672af85dc9dfd60e0/httpx2-2.11.0-py3-none-any.whl", hash = "sha256:c9790f62a327110f52a099f1e2030cbe32f78b28781ad68fe58bfd6f23e73ab0", size = 95043, upload-time = "2026-08-18T08:03:52.004Z" }, + { url = "https://files.pythonhosted.org/packages/d8/9c/6fe8931fd9f381042a9e4c7d5a7b4cbf7016b252bec0c99a49fce42c3326/httpx2-2.13.1-py3-none-any.whl", hash = "sha256:6dff50fabc270ee5fd25d845d0b078ed20564579744d6d962850975996d2f9a4", size = 95597, upload-time = "2026-09-23T07:47:20.995Z" }, ] [[package]] @@ -3359,11 +3364,11 @@ wheels = [ [[package]] name = "pyjwt" -version = "2.13.0" +version = "2.15.1" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/3b/81/58d0ac84e1ef3a3843791d6954d94c0b33d526c75eeb1efbce9d0a4c4077/pyjwt-2.13.0.tar.gz", hash = "sha256:41571c89ca91598c79e8ef18a2d07367d4810fbbd6f637794879baf1b7703423", size = 107515, upload-time = "2026-05-21T19:54:36.618Z" } +sdist = { url = "https://files.pythonhosted.org/packages/43/ea/5194e52748b0da83d71e082d75496eaec6e58f419f5e184786ded517e6a9/pyjwt-2.15.1.tar.gz", hash = "sha256:4f259e80cdfb6b3fc18a7de51fd1ef9ec79652f25019bae68975ca2468a34df8", size = 121252, upload-time = "2026-09-28T18:40:42.598Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/a3/5e/ecf12fdb62546d64385c158514e9b2b671f7832108ef2ecd2020ce0af2d1/pyjwt-2.13.0-py3-none-any.whl", hash = "sha256:66adcc2aff09b3f1bbd95fc1e1577df8ac8723c978552fd43304c8a290ac5728", size = 31274, upload-time = "2026-05-21T19:54:35.362Z" }, + { url = "https://files.pythonhosted.org/packages/50/ca/44de4e75f8aadc457f0634be3b542815078ded46dca30efb960edeecad6e/pyjwt-2.15.1-py3-none-any.whl", hash = "sha256:42d59d631f7768a1028a64c7ff581a9bf7519804daf91fc5b6c56e30eec5e193", size = 33860, upload-time = "2026-09-28T18:40:41.429Z" }, ] [package.optional-dependencies] @@ -4337,19 +4342,19 @@ wheels = [ [[package]] name = "tornado" -version = "6.5.7" +version = "6.5.10" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/64/24/95ec527ad67b76d59299e5465b3935d05e4294b7e0290a3924b7487df30b/tornado-6.5.7.tar.gz", hash = "sha256:66c513a76cda70d53907bc27cf1447557699c2e95aa48ba27a442ff61c3ddfc2", size = 519252, upload-time = "2026-06-08T17:34:51.232Z" } +sdist = { url = "https://files.pythonhosted.org/packages/06/61/53d562a57b28c08eda40b258c0f975e360541943ad7c7bef897a40caafda/tornado-6.5.10.tar.gz", hash = "sha256:a6b1ccd08c04b4a06fb5aeb381be99de5ad1e5375c1785e31d78c880feb57687", size = 537910, upload-time = "2026-09-15T13:47:48.73Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/02/dc/c7043cab6fed8ae159fc1923ce829ada35c4dbd797d408a43858ffaf9639/tornado-6.5.7-cp39-abi3-macosx_10_9_universal2.whl", hash = "sha256:148b2eb15c2c765a50796172c1e499649b35f30d2e3c3d3e15913cfa56bfb163", size = 448543, upload-time = "2026-06-08T17:34:38.052Z" }, - { url = "https://files.pythonhosted.org/packages/92/4f/090b1431e5a43df696feceffc268c5383cc079ecb5f08ce58f917109aafe/tornado-6.5.7-cp39-abi3-macosx_10_9_x86_64.whl", hash = "sha256:9da38de27f1da3b78a966f0dae12b5a1ea9afe72ca805d84ff06508272ddf100", size = 446707, upload-time = "2026-06-08T17:34:39.594Z" }, - { url = "https://files.pythonhosted.org/packages/37/d8/ef374952fd5da67d4463122c2b8e5a96536ec10b4b339254c6dcde81d01c/tornado-6.5.7-cp39-abi3-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:8d759e71906ee783f8867b93bf26a265743da4c1e2f4a018464c1ba019862972", size = 449774, upload-time = "2026-06-08T17:34:41.204Z" }, - { url = "https://files.pythonhosted.org/packages/35/37/d434c73f4c6e014b745b9b37085f34f40c022f007efff3d7fe65991899f3/tornado-6.5.7-cp39-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:8a46347a18f23fb92b396beebe0fb78f61dda0cc302445202c16203d8a18848b", size = 450745, upload-time = "2026-06-08T17:34:42.531Z" }, - { url = "https://files.pythonhosted.org/packages/b6/2b/56b9aff361d7f1ab728a805ec7d7ea835f8807afa9f5cc690ea0e630efb9/tornado-6.5.7-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:7778b30bef919231265e91c69963ce0f49a1e9c07ac900bbe75b19ce2575ba92", size = 450578, upload-time = "2026-06-08T17:34:43.787Z" }, - { url = "https://files.pythonhosted.org/packages/02/30/a7444fb23aa76860a14198fab96ac79f1866b0a6e19e26c4381b0938e50f/tornado-6.5.7-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:e726f0c75da7726eec023aa62751ff8878bd2737e34fbdd33b1ae5897d2200f5", size = 449985, upload-time = "2026-06-08T17:34:45.326Z" }, - { url = "https://files.pythonhosted.org/packages/5c/42/5f0e56c01e8d9d36f4e23f367b85ae6cae0c1ecddd5e6977d8388ad27488/tornado-6.5.7-cp39-abi3-win32.whl", hash = "sha256:f8de3bf12d3efdd0cbe7c8887868198f8a91415e3f29fcf258d9b8eb7b1d9ae4", size = 451047, upload-time = "2026-06-08T17:34:46.784Z" }, - { url = "https://files.pythonhosted.org/packages/c9/a4/b393076ffb21b469eec5b328a0534cf03a3b90bfc6b1f09507cdd075d938/tornado-6.5.7-cp39-abi3-win_amd64.whl", hash = "sha256:de942f843533a039ef9fa3d9c88c7cd8a7c94553fb5ad0154270989b3d99a2c4", size = 451485, upload-time = "2026-06-08T17:34:48.248Z" }, - { url = "https://files.pythonhosted.org/packages/71/2e/7b1c769803121b809112cf9a00681c472eae1d80e32d7ec0e0bd61d0d0e1/tornado-6.5.7-cp39-abi3-win_arm64.whl", hash = "sha256:ff934fce95643af5f11efdae618eaa73d469dc588641e5c8d19295a0c65c4796", size = 450506, upload-time = "2026-06-08T17:34:49.702Z" }, + { url = "https://files.pythonhosted.org/packages/cd/5b/ff5fc58fa2427c30dea74c90053f4fc5eda1e7f3833ed3ecc7147fe2b311/tornado-6.5.10-cp39-abi3-macosx_10_9_universal2.whl", hash = "sha256:9261783640e23258694a9ff0795df430a5a7b0a651d3dd53dd0969ad6be16da7", size = 465883, upload-time = "2026-09-15T13:47:35.463Z" }, + { url = "https://files.pythonhosted.org/packages/ad/f5/cd7be26c34a3315532f3aef5f092465da8f59c334dd439d3c14aaef16461/tornado-6.5.10-cp39-abi3-macosx_10_9_x86_64.whl", hash = "sha256:83e6cf438b106c6b3852d70960967bb1b70c87438050dca0981e4b9aa751a4c1", size = 464046, upload-time = "2026-09-15T13:47:37.178Z" }, + { url = "https://files.pythonhosted.org/packages/60/33/df6d7d04854a58619f8349a51e3edb138324130a7562b0bb21f115bb940f/tornado-6.5.10-cp39-abi3-manylinux1_x86_64.manylinux_2_28_x86_64.manylinux_2_5_x86_64.whl", hash = "sha256:bdf942448169e5336451d0494d7e3d81cfa726d5aa312affdc4682dd62a62f6d", size = 467096, upload-time = "2026-09-15T13:47:38.559Z" }, + { url = "https://files.pythonhosted.org/packages/29/17/cc35dff68272d685cffd8600ffafbd8067e7d05e7348d9f80caddffbbd5f/tornado-6.5.10-cp39-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:69acca6501eed74582b76dbbceee2a91613f54728e3e418346000d7103101676", size = 468067, upload-time = "2026-09-15T13:47:40.085Z" }, + { url = "https://files.pythonhosted.org/packages/c3/01/6e5349b4e1a53a4b4972a6716785e1fe7407f312063c3972690af8ff301b/tornado-6.5.10-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:66aaa3f57d30c6e6becee83ff28055d5930ac724214bde99393eefda83d5e015", size = 467901, upload-time = "2026-09-15T13:47:41.576Z" }, + { url = "https://files.pythonhosted.org/packages/28/5e/b4facf94370dba006819c8d304376f8b9fbec6b935b5e51bf45823a9790b/tornado-6.5.10-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:4bd192b959f9128fb99b8898148070ba4574c9589b78bce42d1851131fe85828", size = 467308, upload-time = "2026-09-15T13:47:43.145Z" }, + { url = "https://files.pythonhosted.org/packages/56/ae/047938e828cafc8eca4c908fafb6588fee944e3af39a0af9d7b602499ae5/tornado-6.5.10-cp39-abi3-win32.whl", hash = "sha256:302eb1e0e3e159314eb591920529fdea80acca92df5510a2cec5bbd4f099ec72", size = 468387, upload-time = "2026-09-15T13:47:44.556Z" }, + { url = "https://files.pythonhosted.org/packages/d8/d4/5901517f05affd752490f6a654ba31b7474664e8dd80bd045a00c220bd88/tornado-6.5.10-cp39-abi3-win_amd64.whl", hash = "sha256:37ae8f150cecfdbf747fc4e12f5e9a97ecd8cf1d4cdb3f119e2de84b11196918", size = 468828, upload-time = "2026-09-15T13:47:45.961Z" }, + { url = "https://files.pythonhosted.org/packages/f3/1a/fd497f3a7f7b74bb04f4b94536b5c9f80742b5d50501fd27977652ddec16/tornado-6.5.10-cp39-abi3-win_arm64.whl", hash = "sha256:ce045d3c298fddd30e89a2777f97039d1b641eb9518ac7b26a4721903539c694", size = 467847, upload-time = "2026-09-15T13:47:47.283Z" }, ] [[package]] @@ -4484,11 +4489,11 @@ wheels = [ [[package]] name = "urllib3" -version = "2.7.0" +version = "2.8.0" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" } +sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" }, + { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" }, ] [[package]]