snomed-utils 0.2.0
Production-grade Python library for SNOMED CT access with pluggable backends (Neo4j, Snowstorm, Snowstorm-Lite)
License: MIT
Requires Python: >=3.10
Keywords: fhir, healthcare, neo4j, snomed, snomed-ct, snowstorm, terminology

# SNOMED CT Utilities

A Python library for working with SNOMED CT terminology, supporting Neo4j, Snowstorm, and Snowstorm-Lite backends.

This library enables powerful clinical terminology operations including concept lookups, Expression Constraint Language (ECL) queries, graph traversal, and relationship analysis. It's designed for building intelligent healthcare applications - from AI agents that need to understand medical concepts and their relationships, to clinical decision support systems, electronic health record integrations, and medical NLP pipelines. Whether you're building RAG systems that need semantic medical knowledge or traditional eHealth applications requiring standardized terminology, this library provides a unified, type-safe interface across multiple SNOMED CT backends.

## Installation

```bash
pip install snomed-utils
```

## Quick Start

### Choose Your Client

The library provides three clients, each connecting to a different backend:

- **Neo4JClient** - Graph database backend (most features, best performance)
- **SnowstormClient** - Full Snowstorm terminology server (ECL native)
- **SnowstormLiteClient** - Lightweight FHIR-based server (basic operations)

### Basic Usage

```python
from snomed_utils import Neo4JClient, SnowstormClient, SnowstormLiteClient

# Neo4j Backend
with Neo4JClient(host="localhost", port=7687, username="neo4j", password="password") as client:
    concept = client.lookup("73211009")  # Diabetes mellitus
    print(f"Concept: {concept.fsn}")
    
    # ECL queries
    result = client.ecl_query("<< 73211009")  # All diabetes types
    print(f"Found {len(result.concepts)} concepts")
    
    # Graph operations
    ancestors = client.get_ancestors("73211009")
    distance = client.distance("73211009", "44054006")

# Snowstorm Backend
with SnowstormClient(url="http://localhost:8080") as client:
    concept = client.lookup("73211009")
    result = client.ecl_query("<< 73211009")

# Snowstorm-Lite Backend
with SnowstormLiteClient(url="http://localhost:8080") as client:
    concept = client.lookup("73211009")
    result = client.ecl_query("<< 73211009")
```

### Async Support

All clients have async variants for better performance:

```python
import asyncio
from snomed_utils import Neo4JClientAsync, SnowstormClientAsync, SnowstormLiteClientAsync

async def main():
    async with Neo4JClientAsync(host="localhost", username="neo4j", password="password") as client:
        await client.connect()
        
        # Batch operations are efficient
        concepts = await client.lookup_batch(["73211009", "44054006"])
        print(f"Retrieved {len(concepts)} concepts")
        
        await client.disconnect()

asyncio.run(main())
```

## Client Reference

### Neo4JClient

**Configuration:**
```python
client = Neo4JClient(
    host="localhost",
    port=7687,
    username="neo4j",
    password="password",
    database="neo4j"
)
```

**Key Methods:**
- `lookup(concept_id)` - Get full concept details
- `lookup_batch(concept_ids)` - Batch concept lookup
- `ecl_query(ecl, limit, offset)` - Execute ECL queries (converted to Cypher)
- `get_ancestors(concept_id)` - Get all ancestors
- `get_descendants(concept_id)` - Get all descendants
- `get_siblings(concept_id)` - Get sibling concepts
- `distance(source_id, target_id)` - Calculate hop distance
- `distance_batch(pairs)` - Batch distance calculation
- `load_rf2(rf2_source, release_type, overwrite)` - Load RF2 zip file
- `run_cypher(query, parameters)` - Execute custom Cypher queries

### SnowstormClient

**Configuration:**
```python
client = SnowstormClient(
    url="http://localhost:8080",
    branch="MAIN",
    admin_username="admin",  # For RF2 imports
    admin_password="admin"
)
```

**Key Methods:**
- `lookup(concept_id)` - Get full concept details
- `lookup_batch(concept_ids)` - Batch concept lookup
- `ecl_query(ecl, limit, offset)` - Execute ECL queries (native)
- `get_ancestors(concept_id)` - Get all ancestors via ECL
- `get_descendants(concept_id)` - Get all descendants via ECL
- `get_siblings(concept_id)` - Get sibling concepts
- `import_rf2(file_path)` - Import RF2 zip file
- `get_normalform(concept_id)` - Get normal form expression

### SnowstormLiteClient

**Configuration:**
```python
client = SnowstormLiteClient(
    url="http://localhost:8080",
    admin_username="admin",  # For RF2 imports
    admin_password="2212"
)
```

**Key Methods:**
- `lookup(concept_id)` - Get basic concept details
- `lookup_batch(concept_ids)` - Batch concept lookup
- `ecl_query(ecl, limit, offset)` - Execute ECL queries via FHIR ValueSet
- `get_ancestors(concept_id)` - Limited ancestor support
- `get_descendants(concept_id)` - Limited descendant support
- `import_rf2(file_path)` - Import RF2 zip file
- `get_normalform(concept_id)` - Get normal form expression

## Loading RF2 Data

### Neo4j Backend

```python
from snomed_utils import Neo4JClient

client = Neo4JClient(host="localhost", username="neo4j", password="password")
client.connect()

# Import RF2 zip file
status = client.import_rf2("/path/to/SnomedCT_InternationalRF2.zip")
print(f"Import operation ID: {status.operation_id}")

client.disconnect()
```

### Snowstorm Backend

```python
from snomed_utils import SnowstormClient

client = SnowstormClient(
    url="http://localhost:8080",
    admin_username="admin",
    admin_password="admin"
)
client.connect()

# Import RF2 via admin API
status = client.import_rf2("/path/to/SnomedCT_InternationalRF2.zip")
print(f"Import operation ID: {status.operation_id}")

# Check status
import_status = client.get_import_status(status.operation_id)
print(f"Status: {import_status.status}")

client.disconnect()
```

### Snowstorm-Lite Backend

```python
from snomed_utils import SnowstormLiteClient

client = SnowstormLiteClient(
    url="http://localhost:8080",
    admin_username="admin",
    admin_password="2212"
)
client.connect()

# Import RF2 via admin API
status = client.import_rf2("/path/to/SnomedCT_InternationalRF2.zip")
print(f"Import operation ID: {status.operation_id}")

client.disconnect()
```

## Deployment Examples

Complete deployment examples with Docker Compose are available in the [`.github/deployment_examples/`](.github/deployment_examples/) directory.

### Neo4j

```bash
cd .github/deployment_examples/neo4j
docker-compose up -d
```

The Neo4j browser will be available at http://localhost:7474 with:
- Username: `neo4j`
- Password: `12345678`
- Bolt port: `7687`

See: [`.github/deployment_examples/neo4j/docker-compose.yml`](.github/deployment_examples/neo4j/docker-compose.yml)

### Snowstorm

Follow the official Snowstorm deployment guide:
- Repository: https://github.com/IHTSDO/snowstorm
- Docker guide: https://github.com/IHTSDO/snowstorm/blob/master/docs/using-docker.md
- Loading data: https://github.com/IHTSDO/snowstorm/blob/master/docs/loading-snomed.md

Quick start:
```bash
git clone https://github.com/IHTSDO/snowstorm.git
cd snowstorm
docker compose up -d
```

See: [`.github/deployment_examples/snowstorm/README.md`](.github/deployment_examples/snowstorm/README.md)

### Snowstorm-Lite

```bash
docker pull snomedinternational/snowstorm-lite:latest
docker run -p 8080:8080 --name=snowstorm-lite \
  -v snowstorm-lite-volume:/app/lucene-index \
  snomedinternational/snowstorm-lite \
  --index.path=lucene-index/data \
  --admin.password=yourAdminPassword
```

Official documentation: https://github.com/IHTSDO/snowstorm-lite

See: [`.github/deployment_examples/snowstorm-lite/README.md`](.github/deployment_examples/snowstorm-lite/README.md)

## Data Models

All clients return consistent Pydantic models:

```python
from snomed_utils import Concept, ECLResult, DistanceResult

# Full concept
concept = client.lookup("73211009")
print(f"ID: {concept.id}")
print(f"FSN: {concept.fsn}")
print(f"PT: {concept.pt}")
print(f"Active: {concept.active}")
print(f"Descriptions: {len(concept.descriptions)}")
print(f"Relationships: {len(concept.relationships)}")

# ECL query results
result = client.ecl_query("<< 73211009")
print(f"Total: {result.total}")
print(f"Concepts: {len(result.concepts)}")

# Distance results (Neo4j only)
distance = client.distance("73211009", "44054006")
print(f"Distance: {distance.distance}")
print(f"Path: {distance.path}")
```

## Examples

### Find All Descendants

```python
with Neo4JClient(host="localhost", username="neo4j", password="password") as client:
    # Get all types of diabetes
    descendants = client.get_descendants("73211009", include_self=True)
    for concept in descendants:
        print(f"{concept.id}: {concept.fsn}")
```

### ECL Query

```python
with SnowstormClient(url="http://localhost:8080") as client:
    # Find all medications containing aspirin
    result = client.ecl_query(
        "<< 763158003 |Medicinal product (product)| : "
        "762949000 |Has precise active ingredient (attribute)| = 387458008 |Aspirin (substance)|",
        limit=1000
    )
    print(f"Found {len(result.concepts)} aspirin products")
```

### Batch Operations

```python
with Neo4JClient(host="localhost", username="neo4j", password="password") as client:
    # Batch lookup
    concept_ids = ["73211009", "44054006", "267038008"]
    concepts = client.lookup_batch(concept_ids)
    
    # Batch distance calculation
    pairs = [("73211009", "44054006"), ("267038008", "73211009")]
    distances = client.distance_batch(pairs)
    
    for dist in distances:
        print(f"Distance from {dist.source_id} to {dist.target_id}: {dist.distance} hops")
```

## License

MIT License - see [LICENSE](LICENSE) file for details.

## Acknowledgments

- SNOMED International for SNOMED CT
- IHTSDO for Snowstorm and Snowstorm-Lite
- Neo4j for the graph database platform