Repository

Main class and utilities for managing FHIR structure definitions.

AbstractRepository

Path: fhircraft.fhir.resources.repository.AbstractRepository

Bases: ABC, Generic[T]

Abstract base class for generic repositories.

Methods:

Name	Description
get	Retrieve a resource by canonical URL and optional version.
add	Add a resource to the repository.
has	Check if a resource exists in the repository.
get_versions	Get all available versions for a canonical URL.
get_latest_version	Get the latest version for a canonical URL.
set_internet_enabled	Enable or disable internet access for this repository.
parse_canonical_url	Parse a canonical URL to extract base URL and version.
format_canonical_url	Format a canonical URL with optional version.

get abstractmethod

get(canonical_url: str, version: Optional[str] = None) -> T

Retrieve a resource by canonical URL and optional version.

Source code in fhircraft/fhir/resources/repository.py

@abstractmethod
def get(self, canonical_url: str, version: Optional[str] = None) -> T:
    """Retrieve a resource by canonical URL and optional version."""
    pass

add abstractmethod

add(resource: T) -> None

Add a resource to the repository.

Source code in fhircraft/fhir/resources/repository.py

@abstractmethod
def add(self, resource: T) -> None:
    """Add a resource to the repository."""
    pass

has abstractmethod

has(canonical_url: str, version: Optional[str] = None) -> bool

Check if a resource exists in the repository.

Source code in fhircraft/fhir/resources/repository.py

@abstractmethod
def has(self, canonical_url: str, version: Optional[str] = None) -> bool:
    """Check if a resource exists in the repository."""
    pass

get_versions abstractmethod

get_versions(canonical_url: str) -> List[str]

Get all available versions for a canonical URL.

Source code in fhircraft/fhir/resources/repository.py

@abstractmethod
def get_versions(self, canonical_url: str) -> List[str]:
    """Get all available versions for a canonical URL."""
    pass

get_latest_version abstractmethod

get_latest_version(canonical_url: str) -> Optional[str]

Get the latest version for a canonical URL.

Source code in fhircraft/fhir/resources/repository.py

@abstractmethod
def get_latest_version(self, canonical_url: str) -> Optional[str]:
    """Get the latest version for a canonical URL."""
    pass

set_internet_enabled abstractmethod

set_internet_enabled(enabled: bool) -> None

Enable or disable internet access for this repository.

Source code in fhircraft/fhir/resources/repository.py

@abstractmethod
def set_internet_enabled(self, enabled: bool) -> None:
    """Enable or disable internet access for this repository."""
    pass

parse_canonical_url staticmethod

parse_canonical_url(canonical_url: str) -> Tuple[str, Optional[str]]

Parse a canonical URL to extract base URL and version.

Source code in fhircraft/fhir/resources/repository.py

@staticmethod
def parse_canonical_url(canonical_url: str) -> Tuple[str, Optional[str]]:
    """Parse a canonical URL to extract base URL and version."""
    if "|" in canonical_url:
        base_url, version = canonical_url.split("|", 1)
        return base_url.strip(), version.strip()
    return canonical_url.strip(), None

format_canonical_url staticmethod

format_canonical_url(base_url: str, version: Optional[str] = None) -> str

Format a canonical URL with optional version.

Source code in fhircraft/fhir/resources/repository.py

@staticmethod
def format_canonical_url(base_url: str, version: Optional[str] = None) -> str:
    """Format a canonical URL with optional version."""
    if version:
        return f"{base_url}|{version}"
    return base_url

CompositeStructureDefinitionRepository

Path: fhircraft.fhir.resources.repository.CompositeStructureDefinitionRepository

CompositeStructureDefinitionRepository(internet_enabled: bool = True, enable_packages: bool = True, registry_base_url: Optional[str] = None, timeout: float = 30.0)

Bases: AbstractRepository[StructureDefinition]

CompositeStructureDefinitionRepository provides a unified interface for managing, retrieving, and caching FHIR StructureDefinition resources from multiple sources, including local storage, FHIR packages, and online repositories.

This repository supports

• Local caching of StructureDefinitions, with version tracking and duplicate prevention.
• Optional integration with FHIR package repositories for bulk loading and management of definitions.
• Online retrieval of StructureDefinitions when enabled, with automatic local caching of downloaded resources.
• Utilities for loading definitions from files, directories, or pre-loaded dictionaries.
• Management of loaded packages, including loading, checking, and removal.
• Version management, including retrieval of all available versions, the latest version, and removal of specific versions.

Methods:

Name	Description
get	Retrieve a StructureDefinition resource by its canonical URL and optional version.
add	Adds a StructureDefinition to the local repository.
has	Check if a resource with the given canonical URL and optional version exists in the repository.
get_versions	Retrieve all available versions for a given canonical URL.
get_latest_version	Retrieve the latest version string for a given FHIR resource canonical URL.
load_from_directory	Loads all JSON structure definitions from the specified directory and adds them to the repository.
load_from_files	Loads FHIR StructureDefinition resources from one or more JSON files and adds them to the repository.
load_from_definitions	Loads FHIR structure definitions from one or more pre-loaded dictionaries.
set_internet_enabled	Enables or disables internet access for the repository and its underlying HTTP repository.
get_loaded_urls	Returns a list of URLs for all FHIR definitions currently loaded in the repository.
get_all_loaded_urls_with_versions	Returns a dictionary mapping each loaded base URL to a list of its available versions.
clear_local_cache	Clears the local cache of definitions and latest versions.
remove_version	Remove a specific version or all versions of a resource identified by its canonical URL.
load_package	Loads a FHIR package into the repository, making its structure definitions available.
get_loaded_packages	Returns a dictionary of loaded FHIR packages.
has_package	Check if a package with the specified name and optional version exists in the package repository.
remove_package	Remove a package from the package repository.
set_registry_base_url	Sets the base URL for the package registry.

Source code in fhircraft/fhir/resources/repository.py

def __init__(
    self,
    internet_enabled: bool = True,
    enable_packages: bool = True,
    registry_base_url: Optional[str] = None,
    timeout: float = 30.0,
):
    # Structure: {base_url: {version: StructureDefinition}}
    self._local_definitions: Dict[str, Dict[str, StructureDefinition]] = {}
    # Track latest versions: {base_url: latest_version}
    self._latest_versions: Dict[str, str] = {}
    self._internet_enabled = internet_enabled
    self._http_repository = HttpStructureDefinitionRepository()

    # Optional package repository
    self._package_repository: Optional[PackageStructureDefinitionRepository] = None
    if enable_packages:
        self._package_repository = PackageStructureDefinitionRepository(
            internet_enabled=internet_enabled,
            registry_base_url=registry_base_url,
            timeout=timeout,
        )

get

get(canonical_url: str, version: Optional[str] = None) -> StructureDefinition

Retrieve a StructureDefinition resource by its canonical URL and optional version.

This method attempts to find the requested StructureDefinition in the following order: 1. Local repository: Checks for the resource in the local cache, optionally by version. 2. Package repository: If available, attempts to retrieve the resource from a package repository. 3. Internet: If enabled, tries to fetch the resource from an online repository.

If the resource is found in the package or internet repository, it is cached locally for future use.

Parameters:

Name	Type	Description	Default
canonical_url	str	The canonical URL of the StructureDefinition to retrieve.	required
version	Optional[str]	The specific version of the StructureDefinition to retrieve. If not provided, the latest version is used if available.	None

Returns:

Name	Type	Description
StructureDefinition	StructureDefinition	The requested StructureDefinition resource.

Raises:

Type	Description
RuntimeError	If the StructureDefinition cannot be found in any repository.

Source code in fhircraft/fhir/resources/repository.py

def get(
    self, canonical_url: str, version: Optional[str] = None
) -> StructureDefinition:
    """
    Retrieve a StructureDefinition resource by its canonical URL and optional version.

    This method attempts to find the requested StructureDefinition in the following order:
    1. Local repository: Checks for the resource in the local cache, optionally by version.
    2. Package repository: If available, attempts to retrieve the resource from a package repository.
    3. Internet: If enabled, tries to fetch the resource from an online repository.

    If the resource is found in the package or internet repository, it is cached locally for future use.

    Args:
        canonical_url (str): The canonical URL of the StructureDefinition to retrieve.
        version (Optional[str], optional): The specific version of the StructureDefinition to retrieve.
            If not provided, the latest version is used if available.

    Returns:
        StructureDefinition: The requested StructureDefinition resource.

    Raises:
        RuntimeError: If the StructureDefinition cannot be found in any repository.
    """
    base_url, parsed_version = self.parse_canonical_url(canonical_url)
    target_version = version or parsed_version

    # First try local repository
    if base_url in self._local_definitions:
        if target_version:
            # Look for specific version
            if target_version in self._local_definitions[base_url]:
                return self._local_definitions[base_url][target_version]
        else:
            # Get latest version if no version specified
            latest_version = self.get_latest_version(base_url)
            if (
                latest_version
                and latest_version in self._local_definitions[base_url]
            ):
                return self._local_definitions[base_url][latest_version]

    # Try package repository if available
    if self._package_repository and self._package_repository.has(
        canonical_url, version
    ):
        try:
            structure_definition = self._package_repository.get(
                canonical_url, version
            )
            # Cache it locally for future use
            self.add(structure_definition)
            return structure_definition
        except RuntimeError:
            # Package repository couldn't find it, continue to internet fallback
            pass

    # Last chance, lookg for canonical resource definition
    # Try to find the resource in the local FHIR definitions bundle if available
    current_file_path = Path(__file__).resolve()
    profiles_path = (
        current_file_path.parent
        / "definitions"
        / get_FHIR_release_from_version(version or "4.0.0")
        / "profiles-resources.json"
    )
    if profiles_path.exists():
        with open(profiles_path, "r", encoding="utf-8") as f:
            data = json.load(f)
            entry = next(
                (
                    entry
                    for entry in data["entry"]
                    if entry["resource"]["url"] == canonical_url
                    and entry["resource"]["resourceType"] == "StructureDefinition"
                ),
                None,
            )
            if entry:
                structure_def = StructureDefinition.model_validate(
                    entry["resource"]
                )
                self.add(structure_def)
                return structure_def

    # Fall back to internet if enabled
    if self._internet_enabled:
        structure_definition = self._http_repository.get(canonical_url, version)
        if structure_definition:
            # Cache it locally for future use
            self.add(structure_definition)
            return structure_definition

    version_info = f" version {target_version}" if target_version else ""
    raise StructureDefinitionNotFoundError(
        f"Structure definition not found for {base_url}{version_info}. Either load it locally, load the appropriate package, or enable internet access to download it."
    )

add

add(resource: StructureDefinition, fail_if_exists: bool = False) -> None

Adds a StructureDefinition to the local repository.

Parameters:

Name	Type	Description	Default
resource	StructureDefinition	The StructureDefinition instance to add. Must have a 'url' field, and a version either in the URL or in the 'version' field.	required
fail_if_exists	bool	If True, raises a ValueError if a StructureDefinition with the same base URL and version already exists in the repository. Defaults to False.	False

Raises:

Type	Description
ValueError	If the StructureDefinition does not have a 'url' field.
ValueError	If the StructureDefinition does not have a version (in the URL or 'version' field).
ValueError	If a duplicate StructureDefinition is added and fail_if_exists is True.

Source code in fhircraft/fhir/resources/repository.py

def add(self, resource: StructureDefinition, fail_if_exists: bool = False) -> None:
    """
    Adds a StructureDefinition to the local repository.

    Args:
        resource (StructureDefinition): The StructureDefinition instance to add.
            Must have a 'url' field, and a version either in the URL or in the 'version' field.
        fail_if_exists (bool, optional): If True, raises a ValueError if a StructureDefinition
            with the same base URL and version already exists in the repository. Defaults to False.

    Raises:
        ValueError: If the StructureDefinition does not have a 'url' field.
        ValueError: If the StructureDefinition does not have a version (in the URL or 'version' field).
        ValueError: If a duplicate StructureDefinition is added and fail_if_exists is True.

    """
    if not resource.url:
        raise ValueError(
            "StructureDefinition must have a 'url' field to be added to the repository."
        )

    base_url, version = self.parse_canonical_url(resource.url)

    # Use the structure definition's version field if no version in URL
    if not version:
        version = resource.version or resource.fhirVersion

    if not version:
        raise ValueError(
            f"StructureDefinition for {base_url} must have a version (either in URL or version field)."
        )

    # Initialize base URL storage if needed
    if base_url not in self._local_definitions:
        self._local_definitions[base_url] = {}

    # Check for duplicates
    if version in self._local_definitions[base_url] and fail_if_exists:
        raise ValueError(
            f"Attempted to load structure definition with duplicated URL {base_url} version {version} in local repository."
        )

    # Store the definition
    self._local_definitions[base_url][version] = resource

    # Update latest version tracking
    self._update_latest_version(base_url, version)

has

has(canonical_url: str, version: Optional[str] = None) -> bool

Check if a resource with the given canonical URL and optional version exists in the repository. This method searches for the resource in the following order: 1. Local storage: Checks if the resource is available locally, optionally matching the specified version. 2. Package repository: If configured, checks if the resource exists in the package repository. 3. HTTP repository: If internet access is enabled, checks if the resource can be found via the HTTP repository. Args: canonical_url (str): The canonical URL of the resource to check. version (Optional[str], optional): The specific version of the resource to check for. Defaults to None. Returns: bool: True if the resource exists in any of the repositories or can be downloaded; False otherwise.

Source code in fhircraft/fhir/resources/repository.py

def has(self, canonical_url: str, version: Optional[str] = None) -> bool:
    """
    Check if a resource with the given canonical URL and optional version exists in the repository.
    This method searches for the resource in the following order:
    1. Local storage: Checks if the resource is available locally, optionally matching the specified version.
    2. Package repository: If configured, checks if the resource exists in the package repository.
    3. HTTP repository: If internet access is enabled, checks if the resource can be found via the HTTP repository.
    Args:
        canonical_url (str): The canonical URL of the resource to check.
        version (Optional[str], optional): The specific version of the resource to check for. Defaults to None.
    Returns:
        bool: True if the resource exists in any of the repositories or can be downloaded; False otherwise.
    """

    base_url, parsed_version = self.parse_canonical_url(canonical_url)
    target_version = version or parsed_version

    # Check local storage
    if base_url in self._local_definitions:
        if target_version:
            if target_version in self._local_definitions[base_url]:
                return True
        else:
            # Has any version locally
            if self._local_definitions[base_url]:
                return True

    # Check package repository if available
    if self._package_repository and self._package_repository.has(
        canonical_url, version
    ):
        return True

    # Check if can be downloaded
    return self._internet_enabled and self._http_repository.has(
        canonical_url, version
    )

get_versions

get_versions(canonical_url: str) -> List[str]

Retrieve all available versions for a given canonical URL. Args: canonical_url (str): The canonical URL of the resource, possibly including a version. Returns: List[str]: A sorted list of version strings available for the specified canonical URL. Versions are sorted using semantic versioning if possible; otherwise, they are sorted lexicographically.

Source code in fhircraft/fhir/resources/repository.py

def get_versions(self, canonical_url: str) -> List[str]:
    """
    Retrieve all available versions for a given canonical URL.
    Args:
        canonical_url (str): The canonical URL of the resource, possibly including a version.
    Returns:
        List[str]: A sorted list of version strings available for the specified canonical URL.
            Versions are sorted using semantic versioning if possible; otherwise, they are sorted lexicographically.
    """

    base_url, _ = self.parse_canonical_url(canonical_url)

    if base_url in self._local_definitions:
        # Sort versions using semantic versioning
        versions = list(self._local_definitions[base_url].keys())
        try:
            return sorted(versions, key=lambda v: version.parse(v))
        except version.InvalidVersion:
            # Fall back to string sorting if not semantic versions
            return sorted(versions)

    return []

get_latest_version

get_latest_version(canonical_url: str) -> Optional[str]

Retrieve the latest version string for a given FHIR resource canonical URL. Args: canonical_url (str): The canonical URL of the FHIR resource, possibly including a version. Returns: Optional[str]: The latest version string associated with the base canonical URL, or None if no version is found.

Source code in fhircraft/fhir/resources/repository.py

def get_latest_version(self, canonical_url: str) -> Optional[str]:
    """
    Retrieve the latest version string for a given FHIR resource canonical URL.
    Args:
        canonical_url (str): The canonical URL of the FHIR resource, possibly including a version.
    Returns:
        Optional[str]: The latest version string associated with the base canonical URL,
                       or None if no version is found.
    """

    base_url, _ = self.parse_canonical_url(canonical_url)
    return self._latest_versions.get(base_url)

load_from_directory

load_from_directory(directory_path: Union[str, Path]) -> None

Loads all JSON structure definitions from the specified directory and adds them to the repository. Args: directory_path (Union[str, Path]): The path to the directory containing JSON structure definition files. Raises: FileNotFoundError: If the specified directory does not exist. RuntimeError: If an error occurs while loading or adding a structure definition from a file.

Source code in fhircraft/fhir/resources/repository.py

def load_from_directory(self, directory_path: Union[str, Path]) -> None:
    """
    Loads all JSON structure definitions from the specified directory and adds them to the repository.
    Args:
        directory_path (Union[str, Path]): The path to the directory containing JSON structure definition files.
    Raises:
        FileNotFoundError: If the specified directory does not exist.
        RuntimeError: If an error occurs while loading or adding a structure definition from a file.
    """

    directory = Path(directory_path)
    if not directory.exists():
        raise FileNotFoundError(f"Directory not found: {directory_path}")

    for file_path in directory.rglob("*.json"):
        try:
            structure_def = self.__load_json_structure_definition(file_path)
            self.add(structure_def)
        except Exception as e:
            raise RuntimeError(
                f"Error loading structure definition from {file_path}: {e}"
            )

load_from_files

load_from_files(*file_paths: Union[str, Path]) -> None

Loads FHIR StructureDefinition resources from one or more JSON files and adds them to the repository. Args: *file_paths (Union[str, Path]): One or more file paths to JSON files containing FHIR StructureDefinition resources. Raises: RuntimeError: If loading or parsing any of the files fails, a RuntimeError is raised with details about the file and the error.

Source code in fhircraft/fhir/resources/repository.py

def load_from_files(self, *file_paths: Union[str, Path]) -> None:
    """
    Loads FHIR StructureDefinition resources from one or more JSON files and adds them to the repository.
    Args:
        *file_paths (Union[str, Path]): One or more file paths to JSON files containing FHIR StructureDefinition resources.
    Raises:
        RuntimeError: If loading or parsing any of the files fails, a RuntimeError is raised with details about the file and the error.
    """

    for file_path in file_paths:
        try:
            structure_def = self.__load_json_structure_definition(Path(file_path))
            self.add(structure_def)
        except Exception as e:
            raise RuntimeError(f"Failed to load {file_path}: {e}")

load_from_definitions

load_from_definitions(*definitions: Dict[str, Any] | StructureDefinition) -> None

Loads FHIR structure definitions from one or more pre-loaded dictionaries. Each dictionary in definitions should represent a FHIR StructureDefinition resource. The method validates each dictionary and adds the resulting StructureDefinition object to the repository. Args: *definitions (Dict[str, Any]): One or more dictionaries representing FHIR StructureDefinition resources. Raises: ValidationError: If any of the provided dictionaries do not conform to the StructureDefinition model.

Source code in fhircraft/fhir/resources/repository.py

def load_from_definitions(
    self, *definitions: Dict[str, Any] | StructureDefinition
) -> None:
    """
    Loads FHIR structure definitions from one or more pre-loaded dictionaries.
    Each dictionary in `definitions` should represent a FHIR StructureDefinition resource.
    The method validates each dictionary and adds the resulting StructureDefinition
    object to the repository.
    Args:
        *definitions (Dict[str, Any]): One or more dictionaries representing FHIR StructureDefinition resources.
    Raises:
        ValidationError: If any of the provided dictionaries do not conform to the StructureDefinition model.
    """

    """Load structure definitions from pre-loaded dictionaries."""
    for structure_def in definitions:
        structure_definition = StructureDefinition.model_validate(structure_def)
        self.add(structure_definition)

set_internet_enabled

set_internet_enabled(enabled: bool) -> None

Enables or disables internet access for the repository and its underlying HTTP repository. Args: enabled (bool): If True, internet access is enabled; if False, it is disabled.

Source code in fhircraft/fhir/resources/repository.py

def set_internet_enabled(self, enabled: bool) -> None:
    """
    Enables or disables internet access for the repository and its underlying HTTP repository.
    Args:
        enabled (bool): If True, internet access is enabled; if False, it is disabled.
    """

    self._internet_enabled = enabled
    self._http_repository.set_internet_enabled(enabled)

get_loaded_urls

get_loaded_urls() -> List[str]

Returns a list of URLs for all FHIR definitions currently loaded in the repository.

Returns:

Type	Description
List[str]	List[str]: A list containing the URLs of the loaded FHIR definitions.

Source code in fhircraft/fhir/resources/repository.py

def get_loaded_urls(self) -> List[str]:
    """
    Returns a list of URLs for all FHIR definitions currently loaded in the repository.

    Returns:
        List[str]: A list containing the URLs of the loaded FHIR definitions.
    """
    return list(self._local_definitions.keys())

get_all_loaded_urls_with_versions

get_all_loaded_urls_with_versions() -> Dict[str, List[str]]

Returns a dictionary mapping each loaded base URL to a list of its available versions.

Returns:

Type	Description
Dict[str, List[str]]	Dict[str, List[str]]: A dictionary where the keys are base URLs (str) and the values are lists of version strings (List[str]) associated with each URL.

Source code in fhircraft/fhir/resources/repository.py

def get_all_loaded_urls_with_versions(self) -> Dict[str, List[str]]:
    """
    Returns a dictionary mapping each loaded base URL to a list of its available versions.

    Returns:
        Dict[str, List[str]]: A dictionary where the keys are base URLs (str) and the values are lists of version strings (List[str]) associated with each URL.
    """
    return {
        base_url: self.get_versions(base_url)
        for base_url in self._local_definitions.keys()
    }

clear_local_cache

clear_local_cache() -> None

Clears the local cache of definitions and latest versions. This method removes all entries from the internal caches used to store local definitions and their latest versions, effectively resetting the local state.

Source code in fhircraft/fhir/resources/repository.py

def clear_local_cache(self) -> None:
    """
    Clears the local cache of definitions and latest versions.
    This method removes all entries from the internal caches used to store local definitions
    and their latest versions, effectively resetting the local state.
    """
    self._local_definitions.clear()
    self._latest_versions.clear()

remove_version

remove_version(canonical_url: str, version: Optional[str] = None) -> None

Remove a specific version or all versions of a resource identified by its canonical URL. Args: canonical_url (str): The canonical URL of the resource, optionally including a version. version (Optional[str], optional): The specific version to remove. If not provided, the version is parsed from the canonical URL if present. If neither is provided, all versions for the base URL are removed. Returns: None Behavior: - If a version is specified (either as an argument or in the canonical URL), removes only that version. - If the removed version was the latest, updates the latest version to the next available one. - If no versions remain for the base URL, cleans up internal data structures. - If no version is specified, removes all versions associated with the base URL.

Source code in fhircraft/fhir/resources/repository.py

def remove_version(self, canonical_url: str, version: Optional[str] = None) -> None:
    """
    Remove a specific version or all versions of a resource identified by its canonical URL.
    Args:
        canonical_url (str): The canonical URL of the resource, optionally including a version.
        version (Optional[str], optional): The specific version to remove. If not provided, the version is parsed from the canonical URL if present.
                                           If neither is provided, all versions for the base URL are removed.
    Returns:
        None
    Behavior:
        - If a version is specified (either as an argument or in the canonical URL), removes only that version.
        - If the removed version was the latest, updates the latest version to the next available one.
        - If no versions remain for the base URL, cleans up internal data structures.
        - If no version is specified, removes all versions associated with the base URL.
    """

    base_url, parsed_version = self.parse_canonical_url(canonical_url)
    target_version = version or parsed_version

    if base_url not in self._local_definitions:
        return

    if target_version:
        # Remove specific version
        self._local_definitions[base_url].pop(target_version, None)

        # Update latest version if we removed it
        if self._latest_versions.get(base_url) == target_version:
            remaining_versions = self.get_versions(base_url)
            if remaining_versions:
                self._latest_versions[base_url] = remaining_versions[
                    -1
                ]  # Last in sorted list
            else:
                self._latest_versions.pop(base_url, None)

        # Clean up empty base URL entries
        if not self._local_definitions[base_url]:
            del self._local_definitions[base_url]
    else:
        # Remove all versions
        del self._local_definitions[base_url]
        self._latest_versions.pop(base_url, None)

load_package

load_package(package_name: str, version: Optional[str] = None) -> None

Loads a FHIR package into the repository, making its structure definitions available. Args: package_name (str): The name of the FHIR package to load. version (Optional[str], optional): The specific version of the package to load. If None, the latest version is used. Raises: RuntimeError: If package support is not enabled for this repository. Side Effects: - Loads the specified package into the internal package repository. - Adds any new structure definitions from the package to the local cache, avoiding duplicates.

Source code in fhircraft/fhir/resources/repository.py

def load_package(self, package_name: str, version: Optional[str] = None) -> None:
    """
    Loads a FHIR package into the repository, making its structure definitions available.
    Args:
        package_name (str): The name of the FHIR package to load.
        version (Optional[str], optional): The specific version of the package to load. If None, the latest version is used.
    Raises:
        RuntimeError: If package support is not enabled for this repository.
    Side Effects:
        - Loads the specified package into the internal package repository.
        - Adds any new structure definitions from the package to the local cache, avoiding duplicates.
    """

    if not self._package_repository:
        raise RuntimeError("Package support is not enabled for this repository")

    # Load the package (this modifies the package repository's internal state)
    self._package_repository.load_package(package_name, version)

    # Get all structure definitions from the package repository
    # and add any that aren't already in our local cache
    loaded_definitions = []
    for url in self._package_repository._local_definitions:
        for ver, structure_def in self._package_repository._local_definitions[
            url
        ].items():
            # Check if we already have this exact version locally
            if (
                url not in self._local_definitions
                or ver not in self._local_definitions[url]
            ):
                self.add(structure_def)
                loaded_definitions.append(structure_def)

get_loaded_packages

get_loaded_packages() -> Dict[str, str]

Returns a dictionary of loaded FHIR packages.

Returns:

Type	Description
Dict[str, str]	Dict[str, str]: A dictionary where the keys are package names and the values are their corresponding versions.

Source code in fhircraft/fhir/resources/repository.py

def get_loaded_packages(self) -> Dict[str, str]:
    """
    Returns a dictionary of loaded FHIR packages.

    Returns:
        Dict[str, str]: A dictionary where the keys are package names and the values are their corresponding versions.
    """

    if not self._package_repository:
        return {}
    return self._package_repository.get_loaded_packages()

has_package

has_package(package_name: str, version: Optional[str] = None) -> bool

Check if a package with the specified name and optional version exists in the package repository. Args: package_name (str): The name of the package to check for. version (Optional[str], optional): The specific version of the package to check. Defaults to None. Returns: bool: True if the package (and version, if specified) exists in the repository, False otherwise.

Source code in fhircraft/fhir/resources/repository.py

def has_package(self, package_name: str, version: Optional[str] = None) -> bool:
    """
    Check if a package with the specified name and optional version exists in the package repository.
    Args:
        package_name (str): The name of the package to check for.
        version (Optional[str], optional): The specific version of the package to check. Defaults to None.
    Returns:
        bool: True if the package (and version, if specified) exists in the repository, False otherwise.
    """

    if not self._package_repository:
        return False
    return self._package_repository.has_package(package_name, version)

remove_package

remove_package(package_name: str, version: Optional[str] = None) -> None

Remove a package from the package repository. Args: package_name (str): The name of the package to remove. version (Optional[str], optional): The specific version of the package to remove. If None, all versions may be removed. Defaults to None. Returns: None

Source code in fhircraft/fhir/resources/repository.py

def remove_package(self, package_name: str, version: Optional[str] = None) -> None:
    """
    Remove a package from the package repository.
    Args:
        package_name (str): The name of the package to remove.
        version (Optional[str], optional): The specific version of the package to remove. If None, all versions may be removed. Defaults to None.
    Returns:
        None
    """

    if not self._package_repository:
        return
    self._package_repository.remove_package(package_name, version)

set_registry_base_url

set_registry_base_url(base_url: str) -> None

Sets the base URL for the package registry. Args: base_url (str): The base URL to be used for the package registry. Raises: RuntimeError: If package support is not enabled for this repository.

Source code in fhircraft/fhir/resources/repository.py

def set_registry_base_url(self, base_url: str) -> None:
    """
    Sets the base URL for the package registry.
    Args:
        base_url (str): The base URL to be used for the package registry.
    Raises:
        RuntimeError: If package support is not enabled for this repository.
    """

    if not self._package_repository:
        raise RuntimeError("Package support is not enabled for this repository")
    self._package_repository.set_registry_base_url(base_url)

HttpStructureDefinitionRepository

Path: fhircraft.fhir.resources.repository.HttpStructureDefinitionRepository

HttpStructureDefinitionRepository()

Bases: AbstractRepository[StructureDefinition]

Repository that downloads structure definitions from the internet.

Methods:

Name	Description
get	Download structure definition from the internet.
add	HTTP repository doesn't support adding definitions.
has	Check if URL can potentially be resolved.
get_versions	HTTP repository can't list versions without downloading.
get_latest_version	HTTP repository can't determine latest version without downloading.
set_internet_enabled	Enable or disable internet access.

Source code in fhircraft/fhir/resources/repository.py

def __init__(self):
    self._internet_enabled = True

get

get(canonical_url: str, version: Optional[str] = None) -> StructureDefinition

Download structure definition from the internet.

Source code in fhircraft/fhir/resources/repository.py

def get(
    self, canonical_url: str, version: Optional[str] = None
) -> StructureDefinition:
    """Download structure definition from the internet."""
    if not self._internet_enabled:
        raise RuntimeError(
            f"Attempted to get {canonical_url} while internet access is disabled. Either enable internet access or use a local repository."
        )

    # Parse URL to handle versioned URLs
    base_url, parsed_version = self.parse_canonical_url(canonical_url)
    target_version = version or parsed_version

    # Format the URL for download (with version if specified)
    download_url = (
        self.format_canonical_url(base_url, target_version)
        if target_version
        else base_url
    )

    try:
        return self.__download_structure_definition(download_url)
    except ValidationError as ve:
        raise ValidationError(
            f"Validation error for structure definition from {download_url}: {ve}"
        )
    except Exception as e:
        raise RuntimeError(
            f"Failed to download structure definition from {download_url}: {e}"
        )

add

add(resource: StructureDefinition) -> None

HTTP repository doesn't support adding definitions.

Source code in fhircraft/fhir/resources/repository.py

def add(self, resource: StructureDefinition) -> None:
    """HTTP repository doesn't support adding definitions."""
    raise NotImplementedError(
        "HttpStructureDefinitionRepository doesn't support adding definitions"
    )

has

has(canonical_url: str, version: Optional[str] = None) -> bool

Check if URL can potentially be resolved.

Source code in fhircraft/fhir/resources/repository.py

def has(self, canonical_url: str, version: Optional[str] = None) -> bool:
    """Check if URL can potentially be resolved."""
    base_url, parsed_version = self.parse_canonical_url(canonical_url)
    return self._internet_enabled and base_url.startswith(("http://", "https://"))

get_versions

get_versions(canonical_url: str) -> List[str]

HTTP repository can't list versions without downloading.

Source code in fhircraft/fhir/resources/repository.py

def get_versions(self, canonical_url: str) -> List[str]:
    """HTTP repository can't list versions without downloading."""
    raise NotImplementedError(
        "HttpStructureDefinitionRepository doesn't support getting versions"
    )

get_latest_version

get_latest_version(canonical_url: str) -> Optional[str]

HTTP repository can't determine latest version without downloading.

Source code in fhircraft/fhir/resources/repository.py

def get_latest_version(self, canonical_url: str) -> Optional[str]:
    """HTTP repository can't determine latest version without downloading."""
    raise NotImplementedError(
        "HttpStructureDefinitionRepository doesn't support getting latest version"
    )

set_internet_enabled

set_internet_enabled(enabled: bool) -> None

Enable or disable internet access.

Source code in fhircraft/fhir/resources/repository.py

def set_internet_enabled(self, enabled: bool) -> None:
    """Enable or disable internet access."""
    self._internet_enabled = enabled

PackageStructureDefinitionRepository

Path: fhircraft.fhir.resources.repository.PackageStructureDefinitionRepository

PackageStructureDefinitionRepository(internet_enabled: bool = True, registry_base_url: Optional[str] = None, timeout: float = 30.0)

Bases: AbstractRepository[StructureDefinition]

Repository that can load FHIR packages from package registries.

Parameters:

Name	Type	Description	Default
internet_enabled	bool	Whether to enable internet access	True
registry_base_url	Optional[str]	Base URL for package registry (defaults to packages.fhir.org)	None
timeout	float	Request timeout in seconds	30.0

Methods:

Name	Description
get	Get structure definition from loaded packages.
add	Add a structure definition to the repository.
has	Check if structure definition exists in loaded packages.
get_versions	Get all available versions for a canonical URL.
get_latest_version	Get the latest version for a canonical URL.
set_internet_enabled	Enable or disable internet access.
load_package	Load a FHIR package from the registry and add all structure definitions.
get_loaded_packages	Get a dictionary of loaded packages and their versions.
has_package	Check if a package is loaded.
remove_package	Remove a loaded package and all its structure definitions.
set_registry_base_url	Change the package registry base URL.
clear_local_cache	Clear all locally cached structure definitions.

Source code in fhircraft/fhir/resources/repository.py

def __init__(
    self,
    internet_enabled: bool = True,
    registry_base_url: Optional[str] = None,
    timeout: float = 30.0,
):
    """
    Initialize the package repository.

    Args:
        internet_enabled: Whether to enable internet access
        registry_base_url: Base URL for package registry (defaults to packages.fhir.org)
        timeout: Request timeout in seconds
    """
    self._internet_enabled = internet_enabled
    self._package_client = FHIRPackageRegistryClient(
        base_url=registry_base_url, timeout=timeout
    )
    # Structure: {base_url: {version: StructureDefinition}}
    self._local_definitions: Dict[str, Dict[str, StructureDefinition]] = {}
    # Track latest versions: {base_url: latest_version}
    self._latest_versions: Dict[str, str] = {}
    # Track loaded packages to avoid duplicate loading
    self._loaded_packages: Dict[str, str] = {}  # {package_name: version}

get

get(canonical_url: str, version: Optional[str] = None) -> StructureDefinition

Get structure definition from loaded packages.

Source code in fhircraft/fhir/resources/repository.py

def get(
    self, canonical_url: str, version: Optional[str] = None
) -> StructureDefinition:
    """Get structure definition from loaded packages."""
    base_url, parsed_version = self.parse_canonical_url(canonical_url)
    target_version = version or parsed_version

    # Check local storage
    if base_url in self._local_definitions:
        if target_version:
            # Look for specific version
            if target_version in self._local_definitions[base_url]:
                return self._local_definitions[base_url][target_version]
        else:
            # Get latest version if no version specified
            latest_version = self.get_latest_version(base_url)
            if (
                latest_version
                and latest_version in self._local_definitions[base_url]
            ):
                return self._local_definitions[base_url][latest_version]

    version_info = f" version {target_version}" if target_version else ""
    raise RuntimeError(
        f"Structure definition not found for {base_url}{version_info}. "
        f"Load the appropriate package first using load_package()."
    )

add

add(resource: StructureDefinition) -> None

Add a structure definition to the repository.

Source code in fhircraft/fhir/resources/repository.py

def add(self, resource: StructureDefinition) -> None:
    """Add a structure definition to the repository."""
    if not resource.url:
        raise ValueError(
            "StructureDefinition must have a 'url' field to be added to the repository."
        )

    base_url, version = self.parse_canonical_url(resource.url)

    # Use the structure definition's version field if no version in URL
    if not version and resource.version:
        version = resource.version

    if not version:
        raise ValueError(
            f"StructureDefinition for {base_url} must have a version (either in URL or version field)."
        )

    # Initialize base URL storage if needed
    if base_url not in self._local_definitions:
        self._local_definitions[base_url] = {}

    # Store the definition
    self._local_definitions[base_url][version] = resource

    # Update latest version tracking
    self._update_latest_version(base_url, version)

has

has(canonical_url: str, version: Optional[str] = None) -> bool

Check if structure definition exists in loaded packages.

Source code in fhircraft/fhir/resources/repository.py

def has(self, canonical_url: str, version: Optional[str] = None) -> bool:
    """Check if structure definition exists in loaded packages."""
    base_url, parsed_version = self.parse_canonical_url(canonical_url)
    target_version = version or parsed_version

    # Check local storage
    if base_url in self._local_definitions:
        if target_version:
            return target_version in self._local_definitions[base_url]
        else:
            # Has any version locally
            return bool(self._local_definitions[base_url])

    return False

get_versions

get_versions(canonical_url: str) -> List[str]

Get all available versions for a canonical URL.

Source code in fhircraft/fhir/resources/repository.py

def get_versions(self, canonical_url: str) -> List[str]:
    """Get all available versions for a canonical URL."""
    base_url, _ = self.parse_canonical_url(canonical_url)

    if base_url in self._local_definitions:
        # Sort versions using semantic versioning
        versions = list(self._local_definitions[base_url].keys())
        try:
            return sorted(versions, key=lambda v: version.parse(v))
        except version.InvalidVersion:
            # Fall back to string sorting if not semantic versions
            return sorted(versions)

    return []

get_latest_version

get_latest_version(canonical_url: str) -> Optional[str]

Get the latest version for a canonical URL.

Source code in fhircraft/fhir/resources/repository.py

def get_latest_version(self, canonical_url: str) -> Optional[str]:
    """Get the latest version for a canonical URL."""
    base_url, _ = self.parse_canonical_url(canonical_url)
    return self._latest_versions.get(base_url)

set_internet_enabled

set_internet_enabled(enabled: bool) -> None

Enable or disable internet access.

Source code in fhircraft/fhir/resources/repository.py

def set_internet_enabled(self, enabled: bool) -> None:
    """Enable or disable internet access."""
    self._internet_enabled = enabled

load_package

load_package(package_name: str, package_version: Optional[str] = None, install_dependencies: bool = True, fail_if_exists: bool = False) -> None

Load a FHIR package from the registry and add all structure definitions.

Parameters:

Name	Type	Description	Default
package_name	str	Name of the package (e.g., "hl7.fhir.us.core")	required
package_version	Optional[str]	Version of the package (defaults to latest)	None
install_dependencies	bool	If True, checks and installs any dependencies of the package	True
fail_if_exists	bool	If True, raise error if package already loaded	False

Raises:

Type	Description
PackageNotFoundError	If package or version not found
FHIRPackageRegistryError	If download fails
RuntimeError	If package processing fails

Source code in fhircraft/fhir/resources/repository.py

def load_package(
    self,
    package_name: str,
    package_version: Optional[str] = None,
    install_dependencies: bool = True,
    fail_if_exists: bool = False,
) -> None:
    """
    Load a FHIR package from the registry and add all structure definitions.

    Args:
        package_name: Name of the package (e.g., "hl7.fhir.us.core")
        package_version: Version of the package (defaults to latest)
        install_dependencies: If True, checks and installs any dependencies of the package
        fail_if_exists: If True, raise error if package already loaded

    Raises:
        PackageNotFoundError: If package or version not found
        FHIRPackageRegistryError: If download fails
        RuntimeError: If package processing fails
    """
    if not self._internet_enabled:
        raise RuntimeError(
            f"Cannot load package {package_name} while internet access is disabled"
        )

    # Determine version to load
    target_version = package_version
    if not target_version:
        try:
            target_version = self._package_client.get_latest_version(package_name)
        except (PackageNotFoundError, FHIRPackageRegistryError) as e:
            raise PackageNotFoundError(
                f"Failed to get latest version for package {package_name}: {e}"
            )

    if not target_version:
        raise PackageNotFoundError(
            f"No latest version found for package {package_name}"
        )

    # Check if already loaded
    package_key = f"{package_name}@{target_version}"
    if package_key in self._loaded_packages and fail_if_exists:
        raise ValueError(f"Package {package_key} is already loaded")

    try:
        # Download and extract package
        result = self._package_client.download_package(
            package_name, target_version, extract=True
        )

        # Ensure we got a TarFile object (should be guaranteed when extract=True)
        if not isinstance(result, tarfile.TarFile):
            raise RuntimeError(
                f"Expected TarFile object but got {type(result)} when downloading package"
            )

        try:
            self._process_package_tar(
                result, package_name, target_version, install_dependencies
            )
        finally:
            result.close()

        # Track loaded package
        self._loaded_packages[package_key] = target_version

    except (PackageNotFoundError, FHIRPackageRegistryError) as e:
        raise e
    except Exception as e:
        raise RuntimeError(
            f"Failed to process package {package_name}@{target_version}: {e}"
        )

get_loaded_packages

get_loaded_packages() -> Dict[str, str]

Get a dictionary of loaded packages and their versions.

Source code in fhircraft/fhir/resources/repository.py

def get_loaded_packages(self) -> Dict[str, str]:
    """Get a dictionary of loaded packages and their versions."""
    return self._loaded_packages.copy()

has_package

has_package(package_name: str, package_version: Optional[str] = None) -> bool

Check if a package is loaded.

Parameters:

Name	Type	Description	Default
package_name	str	Name of the package	required
package_version	Optional[str]	Version of the package (if None, checks any version)	None

Returns:

Type	Description
bool	True if package is loaded

Source code in fhircraft/fhir/resources/repository.py

def has_package(
    self, package_name: str, package_version: Optional[str] = None
) -> bool:
    """
    Check if a package is loaded.

    Args:
        package_name: Name of the package
        package_version: Version of the package (if None, checks any version)

    Returns:
        True if package is loaded
    """
    if package_version:
        return f"{package_name}@{package_version}" in self._loaded_packages
    else:
        return any(
            key.startswith(f"{package_name}@") for key in self._loaded_packages
        )

remove_package

remove_package(package_name: str, package_version: Optional[str] = None) -> None

Remove a loaded package and all its structure definitions.

Parameters:

Name	Type	Description	Default
package_name	str	Name of the package	required
package_version	Optional[str]	Version of the package (if None, removes all versions)	None

Source code in fhircraft/fhir/resources/repository.py

def remove_package(
    self, package_name: str, package_version: Optional[str] = None
) -> None:
    """
    Remove a loaded package and all its structure definitions.

    Args:
        package_name: Name of the package
        package_version: Version of the package (if None, removes all versions)
    """
    if package_version:
        package_key = f"{package_name}@{package_version}"
        if package_key in self._loaded_packages:
            del self._loaded_packages[package_key]
    else:
        # Remove all versions of the package
        keys_to_remove = [
            key
            for key in self._loaded_packages
            if key.startswith(f"{package_name}@")
        ]
        for key in keys_to_remove:
            del self._loaded_packages[key]

set_registry_base_url

set_registry_base_url(base_url: str) -> None

Change the package registry base URL.

Source code in fhircraft/fhir/resources/repository.py

def set_registry_base_url(self, base_url: str) -> None:
    """Change the package registry base URL."""
    self._package_client.base_url = base_url

clear_local_cache

clear_local_cache() -> None

Clear all locally cached structure definitions.

Source code in fhircraft/fhir/resources/repository.py

def clear_local_cache(self) -> None:
    """Clear all locally cached structure definitions."""
    self._local_definitions.clear()
    self._latest_versions.clear()
    self._loaded_packages.clear()

StructureDefinitionNotFoundError

Path: fhircraft.fhir.resources.repository.StructureDefinitionNotFoundError

Bases: FileNotFoundError

Raised when a required structure definition cannot be resolved.

T module-attribute

T = TypeVar('T')

configure_repository

configure_repository(directory: Optional[Union[str, Path]] = None, files: Optional[List[Union[str, Path]]] = None, definitions: Optional[List[Dict[str, Any]]] = None, internet_enabled: bool = True) -> CompositeStructureDefinitionRepository

Configures and returns a CompositeStructureDefinitionRepository by loading structure definitions from a directory, a list of files, or a list of definition dictionaries. Args: directory (Optional[Union[str, Path]]): Path to a directory containing structure definition files to load. files (Optional[List[Union[str, Path]]]): List of file paths to structure definition files to load. definitions (Optional[List[Dict[str, Any]]]): List of structure definition dictionaries to load directly. internet_enabled (bool): Whether to enable internet access for the repository (default is True). Returns: CompositeStructureDefinitionRepository: The configured repository with the loaded structure definitions.

Source code in fhircraft/fhir/resources/repository.py

def configure_repository(
    directory: Optional[Union[str, Path]] = None,
    files: Optional[List[Union[str, Path]]] = None,
    definitions: Optional[List[Dict[str, Any]]] = None,
    internet_enabled: bool = True,
) -> CompositeStructureDefinitionRepository:
    """
    Configures and returns a CompositeStructureDefinitionRepository by loading structure definitions
    from a directory, a list of files, or a list of definition dictionaries.
    Args:
        directory (Optional[Union[str, Path]]): Path to a directory containing structure definition files to load.
        files (Optional[List[Union[str, Path]]]): List of file paths to structure definition files to load.
        definitions (Optional[List[Dict[str, Any]]]): List of structure definition dictionaries to load directly.
        internet_enabled (bool): Whether to enable internet access for the repository (default is True).
    Returns:
        CompositeStructureDefinitionRepository: The configured repository with the loaded structure definitions.
    """
    repo = CompositeStructureDefinitionRepository(internet_enabled=internet_enabled)

    if directory:
        repo.load_from_directory(directory)

    if files:
        repo.load_from_files(*files)

    if definitions:
        repo.load_from_definitions(*definitions)

    return repo