#
# Copyright © 2018-2026 Commonwealth Scientific and Industrial Research
# Organisation (CSIRO) ABN 41 687 119 230.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Pathling context and Spark session construction for the Python API.
Author: John Grimes.
"""
# noinspection PyPackageRequirements
import os
import warnings
from typing import TYPE_CHECKING, Dict, Optional, Sequence
from py4j.java_gateway import JavaObject
from pyspark import SparkContext
from pyspark.sql import Column, DataFrame, SparkSession
from pathling._spark_defaults import (
managed_spark_defaults,
merge_spark_conf,
validate_spark_conf,
)
from pathling.fhir import MimeType
if TYPE_CHECKING:
from .datasource import DataSources
__all__ = ["PathlingContext"]
def _java_map(jvm, entries: Dict[str, str]) -> JavaObject:
"""Converts a Python dict of strings to a Java map.
:param jvm: the Py4J JVM view used to construct the map
:param entries: the entries to copy into the map
:return: a ``java.util.HashMap`` holding the same entries
"""
java_map = jvm.java.util.HashMap()
for key, value in entries.items():
java_map.put(key, value)
return java_map
def _convert_java_value(value):
"""Converts a Java value from Py4J to an equivalent Python type."""
if isinstance(value, (str, int, float, bool)):
return value
# For other types (e.g., Java BigDecimal, Row), convert to string.
return str(value)
def _convert_typed_values(jtyped_values) -> list:
"""Converts a Java list of TypedValue objects to Python dicts.
:param jtyped_values: iterable of Java TypedValue objects with getType() and getValue()
:return: list of dicts with ``type`` and ``value`` keys
"""
results = []
for jtyped_value in jtyped_values:
value = jtyped_value.getValue()
if value is not None:
value = _convert_java_value(value)
results.append({"type": jtyped_value.getType(), "value": value})
return results
def _build_spark_session(extra_configs: Optional[dict] = None) -> SparkSession:
"""Builds a Spark session configured with the Pathling and Delta packages.
The managed package coordinates, Delta SQL extension, and Delta catalog are
sourced from :func:`pathling._spark_defaults.managed_spark_defaults`, the
single source of truth shared with the CLI merge logic. Callers that need to
bring their own session (for example to set driver JVM options before launch)
supply those via ``extra_configs`` rather than re-declaring the package
configuration; a caller entry overrides the matching default for that key.
:param extra_configs: optional additional Spark configuration entries to set
on the builder before the session is created.
:return: a configured :class:`SparkSession`.
"""
builder = SparkSession.builder
# Apply Pathling's managed defaults first, then any caller-supplied entries,
# so that a caller value (for example a quiet-logging option) overrides the
# corresponding default for the same key.
for key, value in managed_spark_defaults().items():
builder = builder.config(key, value)
for key, value in (extra_configs or {}).items():
builder = builder.config(key, value)
return builder.getOrCreate()
[docs]class StorageType:
MEMORY: str = "memory"
DISK: str = "disk"
# noinspection PyProtectedMember
[docs]class PathlingContext:
"""
Main entry point for Pathling API functionality.
Should be instantiated with the :func:`PathlingContext.create` class method.
Example use::
pc = PathlingContext.create(spark)
patient_df = pc.encode(spark.read.text('ndjson_resources'), 'Patient')
patient_df.show()
"""
@property
def spark(self) -> SparkSession:
"""
Returns the SparkSession associated with this context.
"""
return self._spark
[docs] @classmethod
def create(
cls,
spark: Optional[SparkSession] = None,
spark_conf: Optional[Dict[str, str]] = None,
max_nesting_level: Optional[int] = 3,
enable_extensions: Optional[bool] = False,
enabled_open_types: Optional[Sequence[str]] = (
"boolean",
"code",
"date",
"dateTime",
"decimal",
"integer",
"string",
"Coding",
"CodeableConcept",
"Address",
"Identifier",
"Reference",
),
enable_terminology: Optional[bool] = True,
terminology_server_url: Optional[str] = "https://tx.ontoserver.csiro.au/fhir",
terminology_verbose_request_logging: Optional[bool] = False,
terminology_socket_timeout: Optional[int] = 60_000,
max_connections_total: Optional[int] = 32,
max_connections_per_route: Optional[int] = 16,
terminology_retry_enabled: Optional[bool] = True,
terminology_retry_count: Optional[int] = 2,
enable_cache: Optional[bool] = True,
cache_max_entries: Optional[int] = 200_000,
cache_storage_type: Optional[str] = StorageType.MEMORY,
cache_storage_path: Optional[str] = None,
cache_default_expiry: Optional[int] = 600,
cache_override_expiry: Optional[int] = None,
token_endpoint: Optional[str] = None,
enable_auth: Optional[bool] = False,
client_id: Optional[str] = None,
client_secret: Optional[str] = None,
scope: Optional[str] = None,
token_expiry_tolerance: Optional[int] = 120,
accept_language: Optional[str] = None,
terminology_mode: Optional[str] = "server",
terminology_storage_path: Optional[str] = None,
default_snomed_edition: Optional[str] = None,
expansion_cache_size: Optional[int] = 100,
dialect_aliases: Optional[Dict[str, str]] = None,
explain_queries: Optional[bool] = False,
max_unbound_traversal_depth: Optional[int] = 10,
enable_delta=False,
enable_remote_debugging: Optional[bool] = False,
debug_port: Optional[int] = 5005,
debug_suspend: Optional[bool] = True,
) -> "PathlingContext":
"""
Creates a :class:`PathlingContext` with the given configuration options. This should only
be done once within a SparkSession - subsequent calls with different configuration may
produce an error.
If no SparkSession is provided, and there is not one already present in this process - a
new SparkSession will be created.
If a SparkSession is not provided, and one is already running within the current process,
it will be reused - and it is assumed that the Pathling library API JAR is already on the
classpath. If you are running your own cluster, make sure it is on the list of packages.
If a SparkSession is provided, it must already have the Pathling library runtime and
Delta Lake packages on its classpath, for example via the ``spark.jars.packages``
coordinates from :func:`pathling._spark_defaults.managed_spark_defaults`.
:param spark: a pre-configured :class:`SparkSession` instance, use this if you need to
control the way that the session is set up
:param spark_conf: extra Spark configuration entries to apply when Pathling builds a
new SparkSession, e.g. ``{"spark.driver.memory": "8g"}``. Keys must begin with
``spark.`` and values must be strings. The entries are merged with Pathling's
managed defaults with the same protection as the CLI's ``--spark-conf`` flag,
so a managed key such as ``spark.jars.packages`` cannot be silently clobbered.
Cannot be combined with ``spark``, and raises :class:`ValueError` if a Spark
driver JVM is already running in the process (for example after an earlier
session, even a stopped one, or under ``spark-submit``): the driver heap and
classpath are fixed when the JVM launches, so settings such as
``spark.driver.memory`` could no longer take effect.
:param max_nesting_level: controls the maximum depth of nested element data that is encoded
upon import. This affects certain elements within FHIR resources that contain
recursive references, e.g. `QuestionnaireResponse.item
<https://hl7.org/fhir/R4/questionnaireresponse.html>`_.
:param enable_extensions: enables support for FHIR extensions
:param enabled_open_types: the list of types that are encoded within open types,
such as extensions. This default list was taken from the data types that are common
to extensions found in widely-used IGs, such as the US and AU base profiles. In
general, you will get the best query performance by encoding your data with the
shortest possible list.
:param enable_terminology: enables the use of terminology functions
:param terminology_server_url: the endpoint of a FHIR terminology service (R4) that the
server can use to resolve terminology queries. The default server is suitable for
testing purposes only.
:param terminology_verbose_request_logging: setting this option to `True` will enable
additional logging of the details of requests to the terminology service. Note that
logging is subject to the Spark logging level, which you can set using
`SparkContext.setLogLevel`. Verbose request logging is sent to the `DEBUG` logging
level.
:param terminology_socket_timeout: the maximum period (in milliseconds) that the server
should wait for incoming data from the HTTP service
:param max_connections_total: the maximum total number of connections for the client
:param max_connections_per_route: the maximum number of connections per route for the client
:param terminology_retry_enabled: controls whether terminology requests that fail for
possibly transient reasons (network connections, DNS problems) should be retried
:param terminology_retry_count: the number of times to retry failed terminology requests
:param enable_cache: set this to false to disable caching of terminology requests (not
recommended)
:param cache_max_entries: sets the maximum number of entries that will be held in memory
:param cache_storage_type: the type of storage to use for the terminology cache. See
`StorageType`.
:param cache_storage_path: the path on disk to use for the cache, required when
`cache_storage_type` is `disk`
:param cache_default_expiry: the default expiry time for cache entries (in seconds), used
when the server does not provide an expiry value
:param cache_override_expiry: if provided, this value overrides the expiry time provided by
the terminology server
:param enable_auth: enables authentication of requests to the terminology server
:param token_endpoint: an OAuth2 token endpoint for use with the client credentials grant
:param client_id: a client ID for use with the client credentials grant
:param client_secret: a client secret for use with the client credentials grant
:param scope: a scope value for use with the client credentials grant
:param token_expiry_tolerance: the minimum number of seconds that a token should have
before expiry when deciding whether to send it with a terminology request
:param accept_language: the default value of the Accept-Language HTTP header passed to
the terminology server. The value may contain multiple languages, with weighted
preferences as defined in
https://www.rfc-editor.org/rfc/rfc9110.html#name-accept-language. If not provided,
the header is not sent. The server can use the header to return the result in the
preferred language if it is able. The actual behaviour may depend on the server
implementation and the code systems used. In local mode this value selects the
preferred display and designation language.
:param terminology_mode: the terminology evaluation backend, either ``"server"`` (the
default, using a remote FHIR terminology server) or ``"local"`` (evaluating against a
local terminology store with no network access).
:param terminology_storage_path: the location of the local terminology store, required when
``terminology_mode`` is ``"local"``.
:param default_snomed_edition: the SNOMED CT module identifier used to disambiguate an
unversioned SNOMED reference when the local store holds multiple editions.
:param expansion_cache_size: the maximum number of value set expansions cached per executor
in local mode.
:param dialect_aliases: additional dialect tags recognised in local mode when a display
is requested in a particular language, mapping a language tag to the
identifier of the SNOMED CT language reference set that serves it (for example
``{"en-NZ": "271000210107"}``). An entry for a tag that is already recognised
replaces the built-in mapping for it.
:param explain_queries: setting this option to `True` will enable additional logging relating
to the query plan used to execute queries
:param max_unbound_traversal_depth: maximum depth for self-referencing structure traversals
in repeat operations. Controls how deeply nested hierarchical data can be flattened
during projection.
:param enable_delta: enables the use of Delta for storage of FHIR data.
Only supported when no SparkSession is provided.
:param enable_remote_debugging: enables remote debugging for the JVM process.
:param debug_port: the port for the debugger to listen on (default: 5005)
:param debug_suspend: if true, the JVM will suspend until a debugger is attached
:return: a :class:`PathlingContext` instance initialized with the specified configuration
"""
# Validate the terminology mode up front, before paying the Spark cold start.
if terminology_mode not in ("server", "local"):
raise ValueError(
f"terminology_mode must be 'server' or 'local', got '{terminology_mode}'"
)
if terminology_mode == "local" and terminology_storage_path is None:
raise ValueError(
"terminology_storage_path is required when terminology_mode is 'local'"
)
# spark_conf only takes effect when Pathling builds a new SparkSession
# and that build launches the driver JVM: launch-time settings (e.g.
# driver memory, packages) are fixed once the JVM is running.
if spark_conf is not None:
if spark is not None:
raise ValueError(
"spark_conf cannot be used together with 'spark': the supplied "
"SparkSession already exists, so the configuration could not "
"take effect. Omit 'spark' to let Pathling build the session, "
"or apply the configuration to your session yourself."
)
# The gateway outlives SparkSession.stop(), and spark-submit starts
# the JVM before Python runs, advertising it via PYSPARK_GATEWAY_PORT.
if (
SparkContext._gateway is not None
or "PYSPARK_GATEWAY_PORT" in os.environ
):
raise ValueError(
"spark_conf cannot take effect because a Spark driver JVM is "
"already running in this process (started by an earlier "
"SparkSession, a notebook or shell, or spark-submit), so "
"launch-time settings such as spark.driver.memory would be "
"ignored. Call create() with spark_conf before any Spark "
"session is created in a fresh process, or configure Spark "
"at launch (e.g. SPARK_DRIVER_MEMORY or spark-submit "
"--driver-memory)."
)
def _new_spark_session():
extra_configs = {}
# Add remote debugging configuration if enabled.
if enable_remote_debugging:
suspend_option = "y" if debug_suspend else "n"
extra_configs["spark.driver.extraJavaOptions"] = (
f"-agentlib:jdwp=transport=dt_socket,server=y,"
f"suspend={suspend_option},address={debug_port}"
)
if spark_conf is not None:
# Validate the user map and merge it with the managed defaults
# using the same protection as the CLI's --spark-conf flag, so
# a managed key (e.g. spark.jars.packages) cannot be clobbered.
# Applied last so an explicit spark_conf entry wins for the same
# key, mirroring the CLI's user-overlay-last convention.
extra_configs.update(
merge_spark_conf(
validate_spark_conf(spark_conf),
on_warning=warnings.warn,
)
)
return _build_spark_session(extra_configs)
spark = spark or SparkSession.getActiveSession() or _new_spark_session()
jvm = spark._jvm
# Build an encoders configuration object from the provided parameters.
encoders_config = (
jvm.au.csiro.pathling.config.EncodingConfiguration.builder()
.maxNestingLevel(max_nesting_level)
.enableExtensions(enable_extensions)
.openTypes(jvm.java.util.HashSet(enabled_open_types))
.build()
)
# Build a terminology client configuration object from the provided parameters.
client_config = (
jvm.au.csiro.pathling.config.HttpClientConfiguration.builder()
.socketTimeout(terminology_socket_timeout)
.maxConnectionsTotal(max_connections_total)
.maxConnectionsPerRoute(max_connections_per_route)
.retryEnabled(terminology_retry_enabled)
.retryCount(terminology_retry_count)
.build()
)
# Build a terminology cache configuration object from the provided parameters.
cache_storage_type_enum = (
jvm.au.csiro.pathling.config.HttpClientCachingStorageType.fromCode(
cache_storage_type
)
)
cache_config = (
jvm.au.csiro.pathling.config.HttpClientCachingConfiguration.builder()
.enabled(enable_cache)
.maxEntries(cache_max_entries)
.storageType(cache_storage_type_enum)
.storagePath(cache_storage_path)
.defaultExpiry(cache_default_expiry)
.overrideExpiry(cache_override_expiry)
.build()
)
# Build a terminology authentication configuration object from the provided parameters.
auth_config = (
jvm.au.csiro.pathling.config.TerminologyAuthConfiguration.builder()
.enabled(enable_auth)
.tokenEndpoint(token_endpoint)
.clientId(client_id)
.clientSecret(client_secret)
.scope(scope)
.tokenExpiryTolerance(token_expiry_tolerance)
.build()
)
# Select the terminology evaluation mode (validated at the top of this method).
mode = (
jvm.au.csiro.pathling.config.TerminologyMode.LOCAL
if terminology_mode == "local"
else jvm.au.csiro.pathling.config.TerminologyMode.SERVER
)
# Build the local terminology configuration when a storage path is supplied.
local_config = None
if terminology_storage_path is not None:
local_builder = (
jvm.au.csiro.pathling.config.LocalTerminologyConfiguration.builder()
.storagePath(terminology_storage_path)
.expansionCacheSize(expansion_cache_size)
)
if default_snomed_edition is not None:
local_builder = local_builder.defaultSnomedEdition(
default_snomed_edition
)
if dialect_aliases:
local_builder = local_builder.dialectAliases(
_java_map(jvm, dialect_aliases)
)
local_config = local_builder.build()
# Build a terminology configuration object from the provided parameters.
terminology_builder = (
jvm.au.csiro.pathling.config.TerminologyConfiguration.builder()
.enabled(enable_terminology)
.serverUrl(terminology_server_url)
.verboseLogging(terminology_verbose_request_logging)
.client(client_config)
.cache(cache_config)
.authentication(auth_config)
.acceptLanguage(accept_language)
.mode(mode)
)
if local_config is not None:
terminology_builder = terminology_builder.local(local_config)
terminology_config = terminology_builder.build()
# Build a query configuration object from the provided parameters.
query_config = (
jvm.au.csiro.pathling.config.QueryConfiguration.builder()
.explainQueries(explain_queries)
.maxUnboundTraversalDepth(max_unbound_traversal_depth)
.build()
)
jpc: JavaObject = (
jvm.au.csiro.pathling.library.PathlingContext.builder(spark._jsparkSession)
.encodingConfiguration(encoders_config)
.terminologyConfiguration(terminology_config)
.queryConfiguration(query_config)
.build()
)
return PathlingContext(spark, jpc)
def __init__(self, spark: SparkSession, jpc: JavaObject) -> None:
self._spark: SparkSession = spark
self._jpc: JavaObject = jpc
def _wrap_df(self, jdf: JavaObject) -> DataFrame:
#
# Before Spark v3.3 Dataframes were constructs with SQLContext, which was available
# in `_wrapped` attribute of SparkSession.
# Since v3.3 Dataframes are constructed with SparkSession instance directly.
#
return DataFrame(
jdf,
self._spark._wrapped if hasattr(self._spark, "_wrapped") else self._spark,
)
[docs] def version(self):
"""
:return: The version of the Pathling library.
"""
return self._jpc.getVersion()
[docs] def import_snomed(
self,
source: str,
storage_path: str,
edition_uri: Optional[str] = None,
dense_id_order: Optional[str] = None,
default_dialect: Optional[str] = None,
) -> None:
"""
Imports a SNOMED CT RF2 snapshot release into a local terminology store.
:param source: the path to an RF2 release archive or extracted directory, on any filesystem
accessible through the Hadoop FileSystem API
:param storage_path: the terminology store location, created if absent
:param edition_uri: an explicit SNOMED edition/version URI, overriding detection
:param dense_id_order: how dense identifiers are assigned to concepts, either
``"code-order"`` (the default) or ``"pre-order"``. The pre-order makes the runtime
hierarchy index materially smaller, in exchange for identifiers that shift more
between releases
:param default_dialect: the dialect whose preferred synonyms become the stored display of
every concept, given as a dialect tag such as ``"en-GB"``, as a private-use dialect
extension tag, or as a language reference set identifier. When omitted, the dialect is
chosen from the release: the sole language reference set where there is only one, or US
English for the International edition. A release that holds several and is not the
International edition fails the import, naming the candidates
:raises: the mapped JVM ``TerminologyImportException`` if the source is not a valid RF2
snapshot release, or the default dialect can be neither honoured nor derived; the
store is left unmodified
"""
jvm = self._spark._jvm
options = None
if (
edition_uri is not None
or dense_id_order is not None
or default_dialect is not None
):
builder = jvm.au.csiro.pathling.library.terminology.TerminologyImportOptions.builder()
if edition_uri is not None:
builder = builder.editionUri(edition_uri)
if dense_id_order is not None:
builder = builder.denseIdOrder(
jvm.au.csiro.pathling.terminology.store.DenseIdOrder.fromValue(
dense_id_order
)
)
if default_dialect is not None:
builder = builder.defaultDialect(default_dialect)
options = builder.build()
self._jpc.importSnomed(source, storage_path, options)
[docs] def import_fhir_terminology(
self,
source: str,
storage_path: str,
verify_package: bool = True,
package_registry: Optional[str] = None,
) -> None:
"""
Imports FHIR R4 CodeSystem, ValueSet, and ConceptMap resources into a local terminology
store.
A FHIR NPM package (``.tgz``) is checked against the checksum its registry publishes before
anything is written: a package that does not match fails the import and leaves the store
untouched. A package the registry does not describe, or a registry that cannot be reached,
is imported and recorded as unverified. For an offline import, or a package that was never
published to a registry, pass ``verify_package=False``; the import is then recorded as
having skipped verification.
:param source: the path to a JSON file, a directory of JSON files, or a FHIR NPM package
(``.tgz``), on any filesystem accessible through the Hadoop FileSystem API
:param storage_path: the terminology store location, created if absent
:param verify_package: whether a package source is checked against its registry's published
checksum; ``True`` by default
:param package_registry: the FHIR package registry to check a package against; when omitted
the library default (``https://packages.fhir.org``) is used
:raises: the mapped JVM ``TerminologyImportException`` if the source contains no importable
resources or an invalid resource, or the package does not match the registry
checksum; the store is left unmodified
"""
options = None
if not verify_package or package_registry is not None:
jvm = self._spark._jvm
builder = (
jvm.au.csiro.pathling.library.terminology.FhirImportOptions.builder()
)
builder = builder.verifyPackage(verify_package)
if package_registry is not None:
builder = builder.packageRegistry(package_registry)
options = builder.build()
self._jpc.importFhirTerminology(source, storage_path, options)
[docs] def value_set(self, canonical: str, max_members: Optional[int] = None) -> DataFrame:
"""
Resolves a value set to its membership and returns it as a relation that Spark SQL can
join to, as described in the SQL on FHIR `Terminology in SQL
<https://build.fhir.org/ig/HL7/sql-on-fhir/en/terminology-in-sql.html>`_ page.
The relation has one row per member and the columns ``system``, ``version``, ``code``,
``display`` and ``inactive``; ``system`` and ``code`` are never null. The membership is
obtained from the configured terminology layer when this method is called, so every use of
the relation sees the same membership.
Example::
pc.value_set(
"http://example.org/ValueSet/cardiovascular-disease|2026"
).createOrReplaceTempView("cvd_codes")
:param canonical: the canonical URL of the value set, optionally with a ``|version``
suffix
:param max_members: the largest membership accepted; 100,000 when omitted
:return: a :class:`DataFrame` containing the value set relation
:raises: the mapped JVM ``TerminologyRelationException`` if the value set cannot be
resolved, or its membership cannot be determined or exceeds the cap
"""
if max_members is None:
return self._wrap_df(self._jpc.valueSet(canonical))
return self._wrap_df(self._jpc.valueSet(canonical, max_members))
[docs] def value_set_from_json(
self, json: str, max_members: Optional[int] = None
) -> DataFrame:
"""
Returns the membership of a ValueSet supplied as FHIR R4 JSON as a relation with the
columns of :meth:`value_set`.
The resource must carry a ``url``, and content that is not valid FHIR R4 is rejected.
Where the resource carries an ``expansion``, that expansion is the membership and the
terminology layer is not consulted; an incomplete expansion is rejected. Where it carries
only a ``compose``, the configured terminology layer expands it.
:param json: the ValueSet resource as FHIR R4 JSON
:param max_members: the largest membership accepted; 100,000 when omitted
:return: a :class:`DataFrame` containing the value set relation
:raises: the mapped JVM ``TerminologyRelationException`` if the JSON is not a ValueSet, or
its membership cannot be determined or exceeds the cap
"""
if max_members is None:
return self._wrap_df(self._jpc.valueSetFromJson(json))
return self._wrap_df(self._jpc.valueSetFromJson(json, max_members))
[docs] def concept_map(
self, canonical: str, max_mappings: Optional[int] = None
) -> DataFrame:
"""
Resolves a concept map to its mappings and returns them as a relation that Spark SQL can
join to, as described in the SQL on FHIR `Terminology in SQL
<https://build.fhir.org/ig/HL7/sql-on-fhir/en/terminology-in-sql.html>`_ page.
The relation has one row per mapping, plus one row with a null ``target_code`` for each
source code the map states has no mapping, and the columns ``source_system``,
``source_version``, ``source_code``, ``source_display``, ``target_system``,
``target_version``, ``target_code``, ``target_display`` and ``relationship``;
``source_system`` and ``source_code`` are never null. ``relationship`` is an R5
ConceptMapRelationship code, converted from the R4 equivalence.
Example::
pc.concept_map(
"http://example.org/ConceptMap/sct-to-icd10|2026"
).createOrReplaceTempView("sct_to_icd10")
:param canonical: the canonical URL of the concept map, optionally with a ``|version``
suffix
:param max_mappings: the largest number of mappings accepted; 100,000 when omitted
:return: a :class:`DataFrame` containing the concept map relation
:raises: the mapped JVM ``TerminologyRelationException`` if the concept map cannot be
resolved, carries content the relation cannot represent, or exceeds the cap
"""
if max_mappings is None:
return self._wrap_df(self._jpc.conceptMap(canonical))
return self._wrap_df(self._jpc.conceptMap(canonical, max_mappings))
[docs] def concept_map_from_json(
self, json: str, max_mappings: Optional[int] = None
) -> DataFrame:
"""
Returns the mappings of a ConceptMap supplied as FHIR R4 JSON as a relation with the
columns of :meth:`concept_map`, without consulting the terminology layer.
The resource must carry a ``url``, and content that is not valid FHIR R4, such as the R5
``noMap`` and ``relationship`` elements, is rejected.
:param json: the ConceptMap resource as FHIR R4 JSON
:param max_mappings: the largest number of mappings accepted; 100,000 when omitted
:return: a :class:`DataFrame` containing the concept map relation
:raises: the mapped JVM ``TerminologyRelationException`` if the JSON is not a ConceptMap,
carries content the relation cannot represent, or exceeds the cap
"""
if max_mappings is None:
return self._wrap_df(self._jpc.conceptMapFromJson(json))
return self._wrap_df(self._jpc.conceptMapFromJson(json, max_mappings))
[docs] def encode(
self,
df: DataFrame,
resource_name: str,
input_type: Optional[str] = None,
column: Optional[str] = None,
) -> DataFrame:
"""
Takes a dataframe with a string representations of FHIR resources in the given column and
encodes the resources of the given types as Spark dataframe.
:param df: a :class:`DataFrame` containing the resources to encode.
:param resource_name: the name of the FHIR resource to extract (Condition, Observation,
etc.)
:param input_type: the mime type of input string encoding. Defaults to
`application/fhir+json`.
:param column: the column in which the resources to encode are stored. If 'None' then the
input dataframe is assumed to have one column of type string.
:return: a :class:`DataFrame` containing the given type of resources encoded into Spark
columns
"""
return self._wrap_df(
self._jpc.encode(
df._jdf, resource_name, input_type or MimeType.FHIR_JSON, column
)
)
[docs] def encode_bundle(
self,
df: DataFrame,
resource_name: str,
input_type: Optional[str] = None,
column: Optional[str] = None,
) -> DataFrame:
"""
Takes a dataframe with a string representations of FHIR bundles in the given column and
encodes the resources of the given types as Spark dataframe.
:param df: a :class:`DataFrame` containing the bundles with the resources to encode.
:param resource_name: the name of the FHIR resource to extract (Condition, Observation,
etc.)
:param input_type: the MIME type of the input string encoding. Defaults to
`application/fhir+json`.
:param column: the column in which the resources to encode are stored. If 'None' then the
input dataframe is assumed to have one column of type string.
:return: a :class:`DataFrame` containing the given type of resources encoded into Spark
columns
"""
return self._wrap_df(
self._jpc.encodeBundle(
df._jdf, resource_name, input_type or MimeType.FHIR_JSON, column
)
)
@property
def read(self) -> "DataSources":
"""
Provides access to the instance of :class:`DataSource` factory.
"""
from pathling.datasource import DataSources
return DataSources(self)
[docs] def fhirpath_to_column(
self, resource_type: str, fhirpath_expression: str
) -> Column:
"""
Converts a FHIRPath expression to a PySpark Column.
This method takes a FHIRPath expression and returns a PySpark Column that
can be used in DataFrame operations such as ``filter()`` and ``select()``.
Boolean expressions can be used for filtering, while other expressions can
be used for value extraction.
The expression should evaluate to a single value per resource row.
Example usage::
pc = PathlingContext.create(spark)
# Boolean expression for filtering
gender_filter = pc.fhirpath_to_column("Patient", "gender = 'male'")
filtered = patients.filter(gender_filter)
# Value expression for selection
name_col = pc.fhirpath_to_column("Patient", "name.given.first()")
names = patients.select(name_col)
:param resource_type: the FHIR resource type (e.g., "Patient", "Observation")
:param fhirpath_expression: the FHIRPath expression to evaluate
:return: a PySpark Column representing the evaluated expression
:raises: IllegalArgumentException if the resource type is invalid
:raises: Exception if the FHIRPath expression is invalid
"""
jcolumn = self._jpc.fhirPathToColumn(resource_type, fhirpath_expression)
return Column(jcolumn)
[docs] def evaluate_fhirpath(
self,
resource_type: str,
resource_json: str,
fhirpath_expression: str,
context_expression: Optional[str] = None,
variables: Optional[dict] = None,
) -> dict:
"""
Evaluates a FHIRPath expression against a single FHIR resource and returns
materialised typed results.
The resource is encoded into a one-row Spark Dataset internally, and the
existing FHIRPath engine is used to evaluate the expression. Results are
collected and returned as typed values.
Example usage::
pc = PathlingContext.create(spark)
result = pc.evaluate_fhirpath("Patient", patient_json, "name.family")
for value in result["results"]:
print(f"{value['type']}: {value['value']}")
:param resource_type: the FHIR resource type (e.g., "Patient", "Observation")
:param resource_json: the FHIR resource as a JSON string
:param fhirpath_expression: the FHIRPath expression to evaluate
:param context_expression: an optional context expression; if provided, the main
expression is composed with the context expression
:param variables: optional named variables available via %variable syntax, or None
:return: a dict with ``results`` (list of dicts with ``type`` and ``value``
keys), ``expectedReturnType`` (string), and ``traces`` (list of
dicts with ``label`` and ``values`` keys)
:raises: Exception if the expression is invalid or evaluation fails
"""
jresult = self._jpc.evaluateFhirPath(
resource_type,
resource_json,
fhirpath_expression,
context_expression,
variables,
)
# Convert Java FhirPathResult to Python dict.
results = _convert_typed_values(jresult.getResults())
# Convert trace results.
traces = [
{
"label": jtrace.getLabel(),
"values": _convert_typed_values(jtrace.getValues()),
}
for jtrace in jresult.getTraces()
]
return {
"results": results,
"expectedReturnType": jresult.getExpectedReturnType(),
"traces": traces,
}
[docs] def search_to_column(self, resource_type: str, search_expression: str) -> Column:
"""
Converts a FHIR search expression to a boolean filter column.
This method takes a FHIR search query string and returns a PySpark Column
that can be used to filter a DataFrame of FHIR resources. The resulting
Column evaluates to true for resources that match the search criteria.
Example usage::
pc = PathlingContext.create(spark)
# Single parameter
gender_filter = pc.search_to_column("Patient", "gender=male")
# Multiple parameters (AND)
combined_filter = pc.search_to_column("Patient", "gender=male&birthdate=ge1990-01-01")
# Combine with Pythonic operators
filter1 = pc.search_to_column("Patient", "gender=male")
filter2 = pc.search_to_column("Patient", "active=true")
combined = filter1 & filter2
# Apply to DataFrame
patients = datasource.read("Patient")
filtered = patients.filter(combined)
:param resource_type: the FHIR resource type (e.g., "Patient", "Observation")
:param search_expression: URL query string format (e.g., "gender=male&birthdate=ge1990")
:return: a PySpark Column representing the boolean filter condition
:raises: IllegalArgumentException if the search expression contains unknown parameters
"""
jcolumn = self._jpc.searchToColumn(resource_type, search_expression)
return Column(jcolumn)