Overview
Drug interaction checking is critical for patient safety in healthcare applications. This guide demonstrates how to use the OMOPHub API to identify potential drug-drug interactions, contraindications, and clinical warnings using standardized medication vocabularies.Use Case: Automatically detect potential drug interactions in patient medication lists to prevent adverse drug events (ADEs) and improve clinical decision-making.
Business Problem
Healthcare organizations face significant challenges with medication safety:- Patient Safety Risk: Drug interactions cause 125,000 deaths annually in the US
- Clinical Decision Support: Providers need real-time interaction alerts during prescribing
- Medication Reconciliation: Complex interaction checking across multiple medications
- Regulatory Compliance: FDA requirements for interaction checking in EHR systems
- Cost Impact: ADEs cost healthcare system $100B+ annually
Solution Architecture
Implementation Guide
Step 1: Set Up Drug Interaction Service
Copy
from omophub import OMOPHubClient
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum
from datetime import datetime
import logging
import hashlib
class InteractionSeverity(Enum):
CONTRAINDICATED = "contraindicated"
MAJOR = "major"
MODERATE = "moderate"
MINOR = "minor"
UNKNOWN = "unknown"
@dataclass
class DrugConcept:
concept_id: int
concept_name: str
concept_code: str
vocabulary_id: str
ingredients: List[Dict[str, Any]]
@dataclass
class DrugInteraction:
drug1: DrugConcept
drug2: DrugConcept
severity: InteractionSeverity
mechanism: str
clinical_effect: str
management: str
evidence_level: str
class DrugInteractionChecker:
def __init__(self, api_key: str):
self.client = OMOPHubClient(api_key=api_key)
self.logger = logging.getLogger(__name__)
# Interaction relationship types to check
self.interaction_relationships = [
"Contraindicated with",
"May interact with",
"Has drug interaction with",
"May be contraindicated with",
"Metabolized by same enzyme as"
]
def normalize_medication(self, medication_text: str) -> Optional[DrugConcept]:
"""Normalize medication text to RxNorm concept"""
try:
# Search for medication in RxNorm
search_results = self.client.search_concepts({
"query": medication_text,
"vocabularies": ["RxNorm"],
"domains": ["Drug"],
"standard_concepts_only": True,
"limit": 10
})
if not search_results["concepts"]:
# Try broader search if no exact match
search_results = self.client.search_concepts({
"query": medication_text,
"vocabularies": ["RxNorm", "SNOMED"],
"domains": ["Drug"],
"include_invalid": False,
"limit": 5
})
if search_results["concepts"]:
best_match = search_results["concepts"][0]
# Get active ingredients
ingredients = self.get_active_ingredients(best_match["concept_id"])
return DrugConcept(
concept_id=best_match["concept_id"],
concept_name=best_match["concept_name"],
concept_code=best_match["concept_code"],
vocabulary_id=best_match["vocabulary_id"],
ingredients=ingredients
)
except Exception as e:
# Hash medication text to prevent PHI leakage in logs
med_hash = hashlib.sha256(medication_text.encode('utf-8')).hexdigest()[:8]
self.logger.error(f"Error normalizing medication [hash:{med_hash}]: {e}")
return None
def get_active_ingredients(self, drug_concept_id: int) -> List[Dict[str, Any]]:
"""Get active ingredients for a drug concept"""
try:
# Get relationships to find active ingredients
relationships = self.client.get_concept_relationships(
drug_concept_id,
relationship_types=["Has active ingredient", "Contains"]
)
ingredients = []
for rel in relationships["relationships"]:
if rel["relationship_id"] in ["Has active ingredient", "Contains"]:
ingredient_concept = self.client.get_concept(rel["target_concept_id"])
ingredients.append({
"concept_id": ingredient_concept["concept_id"],
"concept_name": ingredient_concept["concept_name"],
"concept_code": ingredient_concept["concept_code"]
})
return ingredients
except Exception as e:
self.logger.error(f"Error getting ingredients for concept {drug_concept_id}: {e}")
return []
Step 2: Detect Drug Interactions
Copy
def check_drug_interactions(self, medications: List[str]) -> List[DrugInteraction]:
"""Check for interactions between multiple medications"""
# Normalize all medications first
normalized_drugs = []
for med_text in medications:
drug = self.normalize_medication(med_text)
if drug:
normalized_drugs.append(drug)
if len(normalized_drugs) < 2:
return []
interactions = []
# Check each pair of medications
for i, drug1 in enumerate(normalized_drugs):
for j, drug2 in enumerate(normalized_drugs[i+1:], i+1):
interaction = self.check_pair_interaction(drug1, drug2)
if interaction:
interactions.append(interaction)
# Sort by severity
severity_order = {
InteractionSeverity.CONTRAINDICATED: 0,
InteractionSeverity.MAJOR: 1,
InteractionSeverity.MODERATE: 2,
InteractionSeverity.MINOR: 3,
InteractionSeverity.UNKNOWN: 4
}
interactions.sort(key=lambda x: severity_order[x.severity])
return interactions
def check_pair_interaction(self, drug1: DrugConcept, drug2: DrugConcept) -> Optional[DrugInteraction]:
"""Check for interaction between two specific drugs"""
try:
# Direct drug-to-drug interaction check
direct_interaction = self.check_direct_interaction(drug1, drug2)
if direct_interaction:
return direct_interaction
# Check ingredient-level interactions
ingredient_interaction = self.check_ingredient_interactions(drug1, drug2)
if ingredient_interaction:
return ingredient_interaction
# Check class-based interactions
class_interaction = self.check_drug_class_interactions(drug1, drug2)
if class_interaction:
return class_interaction
except Exception as e:
self.logger.error(f"Error checking interaction between {drug1.concept_name} and {drug2.concept_name}: {e}")
return None
def check_direct_interaction(self, drug1: DrugConcept, drug2: DrugConcept) -> Optional[DrugInteraction]:
"""Check for direct relationships between two drugs"""
try:
# Get all relationships for drug1
relationships = self.client.get_concept_relationships(
drug1.concept_id,
relationship_types=self.interaction_relationships
)
# Check if drug2 is in the relationships
for rel in relationships["relationships"]:
if rel["target_concept_id"] == drug2.concept_id:
severity = self.determine_severity(rel["relationship_id"])
return DrugInteraction(
drug1=drug1,
drug2=drug2,
severity=severity,
mechanism=self.get_interaction_mechanism(rel),
clinical_effect=self.get_clinical_effect(rel),
management=self.get_management_guidance(rel),
evidence_level=rel.get("evidence_level", "Unknown")
)
except Exception as e:
self.logger.error(f"Error in direct interaction check: {e}")
return None
def check_ingredient_interactions(self, drug1: DrugConcept, drug2: DrugConcept) -> Optional[DrugInteraction]:
"""Check for interactions between active ingredients"""
# Define severity ranking (lower number = more severe)
severity_rank = {
InteractionSeverity.CONTRAINDICATED: 0,
InteractionSeverity.MAJOR: 1,
InteractionSeverity.MODERATE: 2,
InteractionSeverity.MINOR: 3,
InteractionSeverity.UNKNOWN: 4
}
most_severe_interaction = None
best_severity_rank = float('inf')
for ingredient1 in drug1.ingredients:
for ingredient2 in drug2.ingredients:
try:
relationships = self.client.get_concept_relationships(
ingredient1["concept_id"],
relationship_types=self.interaction_relationships
)
for rel in relationships["relationships"]:
if rel["target_concept_id"] == ingredient2["concept_id"]:
severity = self.determine_severity(rel["relationship_id"])
current_rank = severity_rank.get(severity, float('inf'))
# Keep track of most severe interaction found
if current_rank < best_severity_rank:
best_severity_rank = current_rank
most_severe_interaction = DrugInteraction(
drug1=drug1,
drug2=drug2,
severity=severity,
mechanism=f"Interaction between {ingredient1['concept_name']} and {ingredient2['concept_name']}",
clinical_effect=self.get_clinical_effect(rel),
management=self.get_management_guidance(rel),
evidence_level=rel.get("evidence_level", "Unknown")
)
except Exception as e:
self.logger.error(f"Error checking ingredient interaction: {e}")
continue
return most_severe_interaction
def check_drug_class_interactions(self, drug1: DrugConcept, drug2: DrugConcept) -> Optional[DrugInteraction]:
"""Check for interactions between drug classes"""
# TODO: Implement class-based interaction checking
# This would involve:
# 1. Getting drug class hierarchies for both drugs using OMOPHub API
# 2. Checking for known class-to-class interactions (e.g., ACE inhibitors + NSAIDs)
# 3. Mapping class interactions to specific drug pair severity
#
# For now, return None to prevent AttributeError
try:
# Placeholder for class-based lookup
# In a full implementation, you would:
# - Get concept ancestors to find drug classes
# - Query interaction databases for class-level interactions
# - Return appropriate DrugInteraction object if found
pass
except Exception as e:
self.logger.error(f"Error in drug class interaction check: {e}")
return None
def determine_severity(self, relationship_type: str) -> InteractionSeverity:
"""Determine interaction severity based on relationship type"""
severity_mapping = {
"Contraindicated with": InteractionSeverity.CONTRAINDICATED,
"May be contraindicated with": InteractionSeverity.MAJOR,
"Has drug interaction with": InteractionSeverity.MODERATE,
"May interact with": InteractionSeverity.MINOR,
"Metabolized by same enzyme as": InteractionSeverity.MINOR
}
return severity_mapping.get(relationship_type, InteractionSeverity.UNKNOWN)
def get_interaction_mechanism(self, relationship: Dict[str, Any]) -> str:
"""Get mechanism of interaction"""
mechanisms = {
"Contraindicated with": "Pharmacodynamic antagonism or severe adverse reaction risk",
"May interact with": "Potential pharmacokinetic or pharmacodynamic interaction",
"Has drug interaction with": "Established interaction mechanism",
"Metabolized by same enzyme as": "Competitive metabolism leading to altered drug levels"
}
return mechanisms.get(relationship["relationship_id"], "Unknown interaction mechanism")
def get_clinical_effect(self, relationship: Dict[str, Any]) -> str:
"""Get clinical effect description"""
# This would typically come from a clinical database
# For demo, return generic descriptions
effects = {
"Contraindicated with": "Serious adverse events or therapeutic failure",
"May interact with": "Altered therapeutic effect or increased side effects",
"Has drug interaction with": "Modified drug efficacy or toxicity",
"Metabolized by same enzyme as": "Altered plasma concentrations"
}
return effects.get(relationship["relationship_id"], "Potential clinical consequence")
def get_management_guidance(self, relationship: Dict[str, Any]) -> str:
"""Get management recommendations"""
guidance = {
"Contraindicated with": "Avoid combination. Consider alternative medications.",
"May interact with": "Monitor closely. Consider dose adjustment or timing modifications.",
"Has drug interaction with": "Monitor for efficacy and toxicity. Dose adjustment may be needed.",
"Metabolized by same enzyme as": "Monitor drug levels. Stagger administration if possible."
}
return guidance.get(relationship["relationship_id"], "Consult clinical pharmacist or physician.")
Step 3: Generate Clinical Report
Copy
def generate_interaction_report(self, medications: List[str], patient_context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Generate comprehensive drug interaction report"""
# Check all interactions
interactions = self.check_drug_interactions(medications)
# Categorize interactions by severity
categorized_interactions = {
"contraindicated": [],
"major": [],
"moderate": [],
"minor": [],
"unknown": []
}
for interaction in interactions:
severity_key = interaction.severity.value
categorized_interactions[severity_key].append(interaction)
# Generate summary statistics
summary = {
"total_medications": len(medications),
"total_interactions": len(interactions),
"contraindicated_count": len(categorized_interactions["contraindicated"]),
"major_count": len(categorized_interactions["major"]),
"moderate_count": len(categorized_interactions["moderate"]),
"minor_count": len(categorized_interactions["minor"]),
"risk_level": self.assess_overall_risk(interactions)
}
# Generate clinical recommendations
recommendations = self.generate_recommendations(interactions, patient_context)
return {
"patient_context": patient_context or {},
"medications_analyzed": medications,
"interactions": categorized_interactions,
"summary": summary,
"recommendations": recommendations,
"generated_at": datetime.now().isoformat()
}
def assess_overall_risk(self, interactions: List[DrugInteraction]) -> str:
"""Assess overall risk level of medication regimen"""
if any(i.severity == InteractionSeverity.CONTRAINDICATED for i in interactions):
return "HIGH_RISK"
elif any(i.severity == InteractionSeverity.MAJOR for i in interactions):
return "MODERATE_RISK"
elif any(i.severity == InteractionSeverity.MODERATE for i in interactions):
return "LOW_MODERATE_RISK"
elif interactions:
return "LOW_RISK"
else:
return "MINIMAL_RISK"
def generate_recommendations(self, interactions: List[DrugInteraction], patient_context: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Generate actionable clinical recommendations"""
recommendations = []
# Critical interactions requiring immediate action
contraindicated = [i for i in interactions if i.severity == InteractionSeverity.CONTRAINDICATED]
if contraindicated:
recommendations.append({
"priority": "URGENT",
"type": "CONTRAINDICATED_COMBINATION",
"message": f"Found {len(contraindicated)} contraindicated drug combination(s). Immediate review required.",
"actions": [
"Discontinue one or both medications",
"Consider alternative therapeutic options",
"Consult prescribing physician immediately"
]
})
# Major interactions requiring monitoring
major = [i for i in interactions if i.severity == InteractionSeverity.MAJOR]
if major:
recommendations.append({
"priority": "HIGH",
"type": "MAJOR_INTERACTION",
"message": f"Found {len(major)} major drug interaction(s). Close monitoring recommended.",
"actions": [
"Monitor patient for adverse effects",
"Consider dose adjustments",
"Increase frequency of clinical assessments",
"Document interaction awareness in medical record"
]
})
# Patient-specific recommendations
if patient_context:
age = patient_context.get("age")
comorbidities = patient_context.get("comorbidities", [])
if age and age >= 65:
recommendations.append({
"priority": "MEDIUM",
"type": "GERIATRIC_CONSIDERATION",
"message": "Patient is elderly (≥65 years). Increased risk for drug interactions.",
"actions": [
"Consider lower starting doses",
"Monitor renal and hepatic function",
"Review medication list regularly",
"Simplify regimen when possible"
]
})
if "kidney_disease" in comorbidities:
recommendations.append({
"priority": "MEDIUM",
"type": "RENAL_CONSIDERATION",
"message": "Patient has renal impairment. Drug clearance may be affected.",
"actions": [
"Adjust doses for renal function",
"Monitor creatinine and eGFR",
"Avoid nephrotoxic combinations"
]
})
return recommendations
Example Implementation
Sample Medication List
Copy
Patient Medications:
1. Warfarin 5mg daily (anticoagulant)
2. Aspirin 81mg daily (antiplatelet)
3. Simvastatin 40mg nightly (statin)
4. Metformin 1000mg twice daily (diabetes)
5. Lisinopril 10mg daily (ACE inhibitor)
Generated Interaction Report
Copy
# Initialize the checker
checker = DrugInteractionChecker("your_api_key")
# Sample medication list
medications = [
"Warfarin 5mg",
"Aspirin 81mg",
"Simvastatin 40mg",
"Metformin 1000mg",
"Lisinopril 10mg"
]
# Patient context
patient_context = {
"age": 72,
"comorbidities": ["diabetes", "hypertension", "atrial_fibrillation"],
"renal_function": "mild_impairment"
}
# Generate report
report = checker.generate_interaction_report(medications, patient_context)
# Print results
print("=== DRUG INTERACTION REPORT ===")
print(f"Risk Level: {report['summary']['risk_level']}")
print(f"Total Interactions: {report['summary']['total_interactions']}")
print("\n=== MAJOR INTERACTIONS ===")
for interaction in report['interactions']['major']:
print(f"⚠️ {interaction.drug1.concept_name} + {interaction.drug2.concept_name}")
print(f" Mechanism: {interaction.mechanism}")
print(f" Clinical Effect: {interaction.clinical_effect}")
print(f" Management: {interaction.management}")
print()
print("=== RECOMMENDATIONS ===")
for rec in report['recommendations']:
print(f"🏥 [{rec['priority']}] {rec['message']}")
for action in rec['actions']:
print(f" • {action}")
print()
Expected Output
Copy
=== DRUG INTERACTION REPORT ===
Risk Level: MODERATE_RISK
Total Interactions: 3
=== MAJOR INTERACTIONS ===
⚠️ Warfarin + Aspirin
Mechanism: Additive anticoagulant effects
Clinical Effect: Increased bleeding risk
Management: Monitor INR closely. Consider gastroprotection.
=== MODERATE INTERACTIONS ===
⚠️ Simvastatin + Warfarin
Mechanism: CYP3A4 interaction affecting warfarin metabolism
Clinical Effect: Potential INR elevation
Management: Monitor INR when starting/stopping statin.
=== RECOMMENDATIONS ===
🏥 [HIGH] Found 1 major drug interaction(s). Close monitoring recommended.
• Monitor patient for adverse effects
• Consider dose adjustments
• Increase frequency of clinical assessments
🏥 [MEDIUM] Patient is elderly (≥65 years). Increased risk for drug interactions.
• Consider lower starting doses
• Monitor renal and hepatic function
• Review medication list regularly
Integration Patterns
1. EHR Integration with Real-time Alerts
Copy
class EHRInteractionIntegration:
def __init__(self, omophub_client, ehr_client):
self.checker = DrugInteractionChecker(omophub_client.api_key)
self.ehr = ehr_client
def check_new_prescription(self, patient_id: str, new_medication: str) -> Dict[str, Any]:
"""Check interactions when prescribing new medication"""
# Get current medications from EHR
current_meds = self.ehr.get_patient_medications(patient_id)
current_med_names = [med['name'] for med in current_meds if med['status'] == 'active']
# Add new medication to the list
all_medications = current_med_names + [new_medication]
# Get patient context
patient = self.ehr.get_patient(patient_id)
patient_context = {
"age": patient.get('age'),
"comorbidities": patient.get('conditions', []),
"allergies": patient.get('allergies', []),
"renal_function": patient.get('renal_status')
}
# Check interactions
report = self.checker.generate_interaction_report(all_medications, patient_context)
# Generate alerts for EHR
alerts = self.generate_ehr_alerts(report, new_medication)
return {
"interaction_report": report,
"alerts": alerts,
"safe_to_prescribe": report['summary']['risk_level'] not in ['HIGH_RISK'],
"requires_override": report['summary']['contraindicated_count'] > 0
}
def generate_ehr_alerts(self, report: Dict[str, Any], new_medication: str) -> List[Dict[str, Any]]:
"""Generate EHR-compatible alerts"""
alerts = []
# Critical alerts for contraindications
for interaction in report['interactions']['contraindicated']:
if new_medication.lower() in interaction.drug1.concept_name.lower() or \
new_medication.lower() in interaction.drug2.concept_name.lower():
alerts.append({
"level": "ERROR",
"title": "CONTRAINDICATED COMBINATION",
"message": f"Contraindication between {interaction.drug1.concept_name} and {interaction.drug2.concept_name}",
"recommendation": interaction.management,
"requires_override": True,
"blocking": True
})
# Warning alerts for major interactions
for interaction in report['interactions']['major']:
if new_medication.lower() in interaction.drug1.concept_name.lower() or \
new_medication.lower() in interaction.drug2.concept_name.lower():
alerts.append({
"level": "WARNING",
"title": "MAJOR INTERACTION",
"message": f"Major interaction between {interaction.drug1.concept_name} and {interaction.drug2.concept_name}",
"recommendation": interaction.management,
"requires_override": False,
"blocking": False
})
return alerts
2. Pharmacy Integration
Copy
class PharmacyInteractionSystem:
def __init__(self, omophub_api_key: str):
self.checker = DrugInteractionChecker(omophub_api_key)
def validate_prescription(self, prescription_data: Dict[str, Any]) -> Dict[str, Any]:
"""Validate prescription at pharmacy level"""
medications = prescription_data.get('medications', [])
patient_data = prescription_data.get('patient', {})
# Normalize medication names
med_names = [med['name'] for med in medications]
# Check interactions
report = self.checker.generate_interaction_report(med_names, patient_data)
# Pharmacy-specific validation
validation_result = {
"prescription_id": prescription_data.get('id'),
"validation_status": self.determine_validation_status(report),
"pharmacist_notes": self.generate_pharmacist_notes(report),
"patient_counseling_points": self.generate_counseling_points(report),
"dispensing_recommendations": self.generate_dispensing_recommendations(report)
}
return validation_result
def determine_validation_status(self, report: Dict[str, Any]) -> str:
"""Determine if prescription can be dispensed"""
if report['summary']['contraindicated_count'] > 0:
return "REJECTED"
elif report['summary']['major_count'] > 0:
return "REQUIRES_PHARMACIST_REVIEW"
elif report['summary']['moderate_count'] > 0:
return "DISPENSE_WITH_COUNSELING"
else:
return "APPROVED"
def generate_counseling_points(self, report: Dict[str, Any]) -> List[str]:
"""Generate patient counseling points"""
counseling_points = []
# Add counseling for each interaction
all_interactions = (
report['interactions']['major'] +
report['interactions']['moderate']
)
for interaction in all_interactions:
counseling_points.append(
f"Monitor for signs of interaction between {interaction.drug1.concept_name} "
f"and {interaction.drug2.concept_name}. {interaction.clinical_effect}"
)
return counseling_points
Performance Optimization
Caching Strategy for Drug Interactions
Copy
import redis
import json
import hashlib
from typing import Optional
class CachedDrugInteractionChecker(DrugInteractionChecker):
def __init__(self, api_key: str, redis_url: str = "redis://localhost:6379"):
super().__init__(api_key)
self.redis_client = redis.from_url(redis_url)
self.cache_ttl = {
"drug_normalization": 86400, # 24 hours
"interactions": 43200, # 12 hours
"ingredients": 86400 # 24 hours
}
def cache_key(self, prefix: str, *args) -> str:
"""Generate cache key"""
key_data = f"{prefix}:{':'.join(str(arg) for arg in args)}"
return hashlib.md5(key_data.encode()).hexdigest()
def normalize_medication(self, medication_text: str) -> Optional[DrugConcept]:
"""Cached medication normalization"""
cache_key = self.cache_key("drug_norm", medication_text.lower())
# Check cache first
cached_result = self.redis_client.get(cache_key)
if cached_result:
data = json.loads(cached_result)
if data:
return DrugConcept(**data)
return None
# Call parent method
result = super().normalize_medication(medication_text)
# Cache result
cache_data = result.__dict__ if result else None
self.redis_client.setex(
cache_key,
self.cache_ttl["drug_normalization"],
json.dumps(cache_data, default=str)
)
return result
def check_pair_interaction(self, drug1: DrugConcept, drug2: DrugConcept) -> Optional[DrugInteraction]:
"""Cached pair interaction checking"""
# Sort IDs to ensure consistent cache keys regardless of order
id1, id2 = sorted([drug1.concept_id, drug2.concept_id])
cache_key = self.cache_key("interaction", id1, id2)
# Check cache first
cached_result = self.redis_client.get(cache_key)
if cached_result:
data = json.loads(cached_result)
if data:
return DrugInteraction(**data)
return None
# Call parent method
result = super().check_pair_interaction(drug1, drug2)
# Cache result
cache_data = result.__dict__ if result else None
self.redis_client.setex(
cache_key,
self.cache_ttl["interactions"],
json.dumps(cache_data, default=str)
)
return result
Best Practices
1. Error Handling and Fallbacks
Copy
def robust_interaction_check(self, medications: List[str]) -> Dict[str, Any]:
"""Robust interaction checking with multiple fallback strategies"""
results = {
"interactions": [],
"warnings": [],
"errors": []
}
try:
# Primary interaction checking
interactions = self.check_drug_interactions(medications)
results["interactions"] = interactions
except APITimeoutError:
# Fallback to simplified checking
self.logger.warning("API timeout, falling back to simplified interaction checking")
try:
simplified_interactions = self.simplified_interaction_check(medications)
results["interactions"] = simplified_interactions
results["warnings"].append("Simplified interaction checking used due to API timeout")
except Exception as e:
results["errors"].append(f"Both primary and fallback checking failed: {e}")
except APIRateLimitError:
# Handle rate limiting
self.logger.warning("Rate limit exceeded, using cached data only")
cached_interactions = self.get_cached_interactions(medications)
results["interactions"] = cached_interactions
results["warnings"].append("Using cached data only due to rate limiting")
except Exception as e:
# Log error but don't fail completely
results["errors"].append(f"Interaction checking failed: {e}")
results["warnings"].append("Manual review required - automated checking unavailable")
return results
def simplified_interaction_check(self, medications: List[str]) -> List[DrugInteraction]:
"""Simplified interaction checking using local rules"""
# Implement basic drug interaction rules
# This would typically use a local database of common interactions
interactions = []
# Example: Check for common dangerous combinations
dangerous_combinations = [
("warfarin", "aspirin"),
("warfarin", "ibuprofen"),
("metformin", "iodinated contrast")
]
med_lower = [med.lower() for med in medications]
for drug1, drug2 in dangerous_combinations:
if any(drug1 in med for med in med_lower) and any(drug2 in med for med in med_lower):
# Create basic interaction object
interactions.append(self.create_basic_interaction(drug1, drug2))
return interactions
2. Security and Compliance
Copy
def secure_interaction_check(self, medications: List[str], user_id: str, audit_context: Dict[str, Any]) -> Dict[str, Any]:
"""HIPAA-compliant interaction checking with audit logging"""
# Audit log entry
audit_entry = {
"timestamp": datetime.now().isoformat(),
"user_id": user_id,
"action": "drug_interaction_check",
"medication_count": len(medications),
"patient_id": audit_context.get('patient_id'),
"session_id": audit_context.get('session_id')
}
try:
# Sanitize medication data for logging
med_hashes = [hashlib.sha256(med.encode()).hexdigest()[:8] for med in medications]
audit_entry["medication_hashes"] = med_hashes
# Perform interaction checking
report = self.generate_interaction_report(medications, audit_context.get('patient_context'))
# Log success
audit_entry["status"] = "success"
audit_entry["interactions_found"] = report['summary']['total_interactions']
audit_entry["risk_level"] = report['summary']['risk_level']
# Remove sensitive data from report if needed
sanitized_report = self.sanitize_report_for_logging(report)
return sanitized_report
except Exception as e:
audit_entry["status"] = "error"
audit_entry["error_message"] = str(e)
raise
finally:
# Always log audit entry
self.audit_logger.log_interaction_check(audit_entry)
def sanitize_report_for_logging(self, report: Dict[str, Any]) -> Dict[str, Any]:
"""Remove sensitive patient data from report for logging"""
sanitized = report.copy()
# Remove sensitive patient context
if 'patient_context' in sanitized:
sanitized['patient_context'] = {
'age_range': self.age_to_range(sanitized['patient_context'].get('age')),
'has_comorbidities': len(sanitized['patient_context'].get('comorbidities', [])) > 0
}
# Keep interaction data but hash medication names
sanitized['medications_analyzed'] = [
hashlib.sha256(med.encode()).hexdigest()[:8]
for med in sanitized['medications_analyzed']
]
return sanitized