Skip to content

Capability Facade

Intent

Provide a simplified interface for querying and managing FHIR server capabilities and configuration, abstracting the complexity of CapabilityStatement parsing.

Forces

  • Endpoint Capability Variation: FHIR servers implement different subsets of the specification, leading to compatibility issues.

Structure

The Capability Facade pattern wraps FHIR CapabilityStatement resources and provides a simplified query interface for capability discovery and matching.

Capability Facade Architecture

Key Components

CapabilityFacade

Main interface for capability queries

CapabilityCache

Caches parsed capability information

CapabilityParser

Parses CapabilityStatement and SMART configuration

QueryEvaluator

Evaluates capability queries against cached data

AggregateView

Combines capabilities from multiple servers

Behavior

Capability Query Flow

The following sequence shows how capability queries are processed:

Capability Facade Sequence

Query Types

  • Resource Support
  • Operation Support
  • Search Parameters
  • Profile Support
  • Security Features

Implementation Considerations

Facade Implementation

Wraps FHIR CapabilityStatement and SMART configuration to provide simplified capability queries with caching and indexing.

Facade Implementation
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
import json

@dataclass
class FHIRCapability:
    """Represents a FHIR server capability"""
    resource_type: str
    interactions: List[str]
    search_params: List[str]
    profiles: List[str]
    operations: List[str]

class CapabilityFacade:
    """
    Provides a simplified interface to FHIR CapabilityStatement
    and SMART configuration for capability queries.
    """

    def __init__(self, capability_statement: Dict, smart_config: Dict = None):
        self.capability_statement = capability_statement
        self.smart_config = smart_config or {}
        self._resource_cache = {}
        self._build_indexes()

    def _build_indexes(self):
        """Build internal indexes for fast capability lookup."""
        self._resource_cache = {}
        self._operation_cache = set()
        self._profile_cache = set()

        for rest in self.capability_statement.get('rest', []):
            # Index system-level operations
            for op in rest.get('operation', []):
                self._operation_cache.add(op.get('name'))

            # Index resources and their capabilities
            for resource in rest.get('resource', []):
                res_type = resource.get('type')
                self._resource_cache[res_type] = FHIRCapability(
                    resource_type=res_type,
                    interactions=[i.get('code') for i in resource.get('interaction', [])],
                    search_params=[s.get('name') for s in resource.get('searchParam', [])],
                    profiles=resource.get('supportedProfile', []),
                    operations=[o.get('name') for o in resource.get('operation', [])]
                )

                # Add profiles to global cache
                self._profile_cache.update(resource.get('supportedProfile', []))

    # ========== Resource Queries ==========

    def supports_resource(self, resource_type: str) -> bool:
        """Check if server supports a resource type."""
        return resource_type in self._resource_cache

    def get_supported_resources(self) -> List[str]:
        """Get list of all supported resource types."""
        return list(self._resource_cache.keys())

    def get_resource_capability(self, resource_type: str) -> Optional[FHIRCapability]:
        """Get full capability information for a resource type."""
        return self._resource_cache.get(resource_type)

    # ========== Interaction Queries ==========

    def supports_interaction(self, resource_type: str, interaction: str) -> bool:
        """Check if specific interaction is supported for resource."""
        capability = self._resource_cache.get(resource_type)
        if not capability:
            return False
        return interaction in capability.interactions

    def can_read(self, resource_type: str) -> bool:
        """Check if resource supports read interaction."""
        return self.supports_interaction(resource_type, 'read')

    def can_search(self, resource_type: str) -> bool:
        """Check if resource supports search-type interaction."""
        return self.supports_interaction(resource_type, 'search-type')

    def can_create(self, resource_type: str) -> bool:
        """Check if resource supports create interaction."""
        return self.supports_interaction(resource_type, 'create')

    def can_update(self, resource_type: str) -> bool:
        """Check if resource supports update interaction."""
        return self.supports_interaction(resource_type, 'update')

    def can_delete(self, resource_type: str) -> bool:
        """Check if resource supports delete interaction."""
        return self.supports_interaction(resource_type, 'delete')

    # ========== Search Parameter Queries ==========

    def get_search_params(self, resource_type: str) -> List[str]:
        """Get supported search parameters for resource."""
        capability = self._resource_cache.get(resource_type)
        return capability.search_params if capability else []

    def supports_search_param(self, resource_type: str, param: str) -> bool:
        """Check if specific search parameter is supported."""
        return param in self.get_search_params(resource_type)

    def supports_chained_search(self) -> bool:
        """Check if server supports chained search parameters."""
        for rest in self.capability_statement.get('rest', []):
            if rest.get('searchParam'):
                # Look for chained parameter indicators
                return True
        return False

    # ========== Operation Queries ==========

    def supports_operation(self, operation: str, resource_type: str = None) -> bool:
        """
        Check if operation is supported.
        If resource_type provided, check resource-level operation.
        Otherwise check system-level operation.
        """
        if resource_type:
            capability = self._resource_cache.get(resource_type)
            if capability and operation in capability.operations:
                return True

        return operation in self._operation_cache

    def get_supported_operations(self, resource_type: str = None) -> List[str]:
        """Get list of supported operations."""
        if resource_type:
            capability = self._resource_cache.get(resource_type)
            return capability.operations if capability else []
        return list(self._operation_cache)

    def supports_bulk_export(self) -> bool:
        """Check if server supports FHIR Bulk Data Export."""
        return self.supports_operation('export')

    def supports_everything(self, resource_type: str = 'Patient') -> bool:
        """Check if server supports $everything operation."""
        return self.supports_operation('everything', resource_type)

    # ========== Profile Queries ==========

    def supports_profile(self, profile_url: str) -> bool:
        """Check if server supports a specific profile."""
        return profile_url in self._profile_cache

    def get_profiles_for_resource(self, resource_type: str) -> List[str]:
        """Get supported profiles for a resource type."""
        capability = self._resource_cache.get(resource_type)
        return capability.profiles if capability else []

    def supports_us_core(self) -> bool:
        """Check if server supports US Core profiles."""
        us_core_prefix = "http://hl7.org/fhir/us/core"
        return any(p.startswith(us_core_prefix) for p in self._profile_cache)

    # ========== SMART Queries ==========

    def get_smart_capabilities(self) -> List[str]:
        """Get SMART on FHIR capabilities."""
        return self.smart_config.get('capabilities', [])

    def supports_smart_launch(self) -> bool:
        """Check if server supports SMART app launch."""
        caps = self.get_smart_capabilities()
        return 'launch-ehr' in caps or 'launch-standalone' in caps

    def get_authorization_endpoint(self) -> Optional[str]:
        """Get OAuth2 authorization endpoint."""
        return self.smart_config.get('authorization_endpoint')

    def get_token_endpoint(self) -> Optional[str]:
        """Get OAuth2 token endpoint."""
        return self.smart_config.get('token_endpoint')

    # ========== Composite Queries ==========

    def evaluate_requirements(self, requirements: Dict) -> Dict:
        """
        Evaluate a set of requirements against capabilities.
        Returns match assessment.
        """
        results = {
            'supported': [],
            'unsupported': [],
            'partial': []
        }

        # Check required resources
        for resource in requirements.get('resources', []):
            if self.supports_resource(resource):
                results['supported'].append(f"Resource: {resource}")
            else:
                results['unsupported'].append(f"Resource: {resource}")

        # Check required operations
        for operation in requirements.get('operations', []):
            if self.supports_operation(operation):
                results['supported'].append(f"Operation: {operation}")
            else:
                results['unsupported'].append(f"Operation: {operation}")

        # Check required profiles
        for profile in requirements.get('profiles', []):
            if self.supports_profile(profile):
                results['supported'].append(f"Profile: {profile}")
            else:
                results['unsupported'].append(f"Profile: {profile}")

        return results

Query Examples

Demonstrates common capability query patterns including resource support, operation availability, profile compliance, and SMART capabilities.

Query Examples
"""
Capability Facade - Query Examples

Demonstrates common capability query patterns using the CapabilityFacade.
"""

from capability_facade import CapabilityFacade

# Example: Load capability statement and create facade
capability_statement = {
    "resourceType": "CapabilityStatement",
    "rest": [{
        "mode": "server",
        "resource": [
            {
                "type": "Patient",
                "interaction": [
                    {"code": "read"},
                    {"code": "search-type"},
                    {"code": "create"},
                    {"code": "update"}
                ],
                "searchParam": [
                    {"name": "identifier"},
                    {"name": "name"},
                    {"name": "birthdate"},
                    {"name": "gender"}
                ],
                "supportedProfile": [
                    "http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient"
                ],
                "operation": [
                    {"name": "everything"}
                ]
            },
            {
                "type": "Observation",
                "interaction": [
                    {"code": "read"},
                    {"code": "search-type"}
                ],
                "searchParam": [
                    {"name": "patient"},
                    {"name": "category"},
                    {"name": "code"},
                    {"name": "date"}
                ],
                "supportedProfile": [
                    "http://hl7.org/fhir/us/core/StructureDefinition/us-core-observation-lab"
                ]
            }
        ],
        "operation": [
            {"name": "export"}
        ]
    }]
}

smart_config = {
    "authorization_endpoint": "https://auth.example.org/authorize",
    "token_endpoint": "https://auth.example.org/token",
    "capabilities": [
        "launch-ehr",
        "launch-standalone",
        "client-public",
        "sso-openid-connect",
        "permission-patient",
        "permission-user"
    ]
}

# Create facade
facade = CapabilityFacade(capability_statement, smart_config)

# ========== Query Examples ==========

# 1. Check resource support
print("=== Resource Support ===")
print(f"Supports Patient: {facade.supports_resource('Patient')}")
print(f"Supports Condition: {facade.supports_resource('Condition')}")
print(f"All resources: {facade.get_supported_resources()}")

# 2. Check interactions
print("\n=== Interaction Support ===")
print(f"Can read Patient: {facade.can_read('Patient')}")
print(f"Can create Patient: {facade.can_create('Patient')}")
print(f"Can delete Patient: {facade.can_delete('Patient')}")
print(f"Can search Observation: {facade.can_search('Observation')}")

# 3. Check search parameters
print("\n=== Search Parameters ===")
print(f"Patient search params: {facade.get_search_params('Patient')}")
print(f"Supports 'identifier': {facade.supports_search_param('Patient', 'identifier')}")
print(f"Supports 'address': {facade.supports_search_param('Patient', 'address')}")

# 4. Check operations
print("\n=== Operations ===")
print(f"Supports bulk export: {facade.supports_bulk_export()}")
print(f"Supports Patient/$everything: {facade.supports_everything('Patient')}")
print(f"System operations: {facade.get_supported_operations()}")

# 5. Check profiles
print("\n=== Profile Support ===")
print(f"Supports US Core: {facade.supports_us_core()}")
print(f"Patient profiles: {facade.get_profiles_for_resource('Patient')}")

# 6. Check SMART capabilities
print("\n=== SMART Capabilities ===")
print(f"Supports SMART launch: {facade.supports_smart_launch()}")
print(f"Auth endpoint: {facade.get_authorization_endpoint()}")
print(f"SMART caps: {facade.get_smart_capabilities()}")

# 7. Evaluate requirements
print("\n=== Requirements Evaluation ===")
requirements = {
    "resources": ["Patient", "Observation", "Condition"],
    "operations": ["export", "validate"],
    "profiles": [
        "http://hl7.org/fhir/us/core/StructureDefinition/us-core-patient"
    ]
}
results = facade.evaluate_requirements(requirements)
print(f"Supported: {results['supported']}")
print(f"Unsupported: {results['unsupported']}")

  • Broker: Broker uses Capability Facade to determine endpoint routing decisions
  • Naming and Trading Service: Naming/Trading uses Capability Facade for capability-based service matching
  • Legacy Adapter: Legacy Adapter uses Capability Facade to expose legacy capabilities as FHIR

Benefits

  • Simplified Queries: Easy to check capabilities without parsing CapabilityStatement
  • Caching: Capability information cached for performance
  • Aggregation: Combine capabilities from multiple servers
  • Abstraction: Hide complexity of capability discovery
  • Dynamic Updates: Track capability changes over time

Trade-offs

  • Staleness: Cached capabilities may become outdated
  • Overhead: Additional component in request path
  • Completeness: Facade may not expose all capability details
  • Maintenance: Must keep up with CapabilityStatement changes

References

Background Refresh

Implement background capability refresh to keep cache current without adding latency to capability queries. Consider webhooks or polling based on server support.