norm_toolkit 2.0.2

ClickHouse backend

The DuckDB builder remains the source of truth. Build a DuckDB file with build_merged_duckdb, then upload its canonical tables into ClickHouse:

uv run python scripts/upload_clickhouse.py data/dbs_final/SmallMolecule.duckdb --database normalization

The upload shows a progress bar for each copied table; pass --no-progress to silence it.

Connection settings are read from .env with python-dotenv and use the official clickhouse-connect client. Set CH_HTTP, for example http://host:8123/normalization; CH_USER and CH_PASSWORD may be supplied separately and override URL credentials.

Use the ClickHouse backend from Python:

import asyncio

from norm_toolkit import ClickHouseNormalizer


async def main():
    normalizer = await ClickHouseNormalizer.create(database="normalization")
    result = await normalizer.normalize(["aspirin"], top_k=5)
    print(result)

    # Walk the SNOMEDCT_US Disease (disorder) hierarchy. Each ancestor is
    # returned once at its shortest graph distance; direct parents have depth 1.
    broader = await normalizer.get_broader_concepts(
        "UMLS:C0006142",
        max_depth=None,
        max_size=2000,
    )
    print([(concept.identifier, concept.depth) for concept in broader])

    await normalizer.aclose()


asyncio.run(main())

Disease hierarchy data is a graph rather than a strict tree. Upward traversal uses only SNOMEDCT_US is-a edges beneath Disease (disorder); it does not cross into other UMLS source vocabularies. Traversals deduplicate diamonds, terminate cycles, and keep the shortest depth for concepts reached by multiple paths. Use max_depth=None to continue to the disease root.

Normalizer instances keep a 10,000-item LRU cache. Repeated items skip string normalization, database lookup, enrichment, and hierarchy expansion; misses in a mixed batch are queried together and added to the cache. Synonyms and all result-affecting options are part of the cache key.

normalizer = await ClickHouseNormalizer.create(
    database="normalization",
    normalization_cache_size=50_000,  # 0 disables; None is unbounded
)

print(normalizer.normalization_cache_info())
normalizer.clear_normalization_cache()  # also use after refreshing backing tables

You can also pass a DSN in code:

normalizer = await ClickHouseNormalizer.create(
    dsn="http://host:8123/normalization",
    database="normalization",
)