fhirpath.engine package

Submodules

fhirpath.engine.base module

class fhirpath.engine.base.Engine(fhir_release, conn_factory, dialect_factory)[source]

Bases: abc.ABC

Idea: # 1.) https://docs.sqlalchemy.org/en/13/core/ # connections.html#sqlalchemy.engine.Engine.connect 2.) https://docs.sqlalchemy.org/en/13/core/ connections.html#sqlalchemy.engine.Connection 3.) Dialect could have raw connection, query compiler 4.) Engine would have execute and result processing through provider, yes provider!

before_execute(query)[source]

Hook: before execution of query

create_connection(factory)[source]
create_dialect(factory)[source]
classmethod is_async()[source]
class fhirpath.engine.base.EngineResult(header: fhirpath.engine.base.EngineResultHeader, body: fhirpath.engine.base.EngineResultBody)[source]

Bases: object

body: fhirpath.engine.base.EngineResultBody
extract_ids() → Dict[str, List[str]][source]
extract_references(search_param: fhirpath.fhirspec.spec.SearchParameter) → Dict[str, List[str]][source]

Takes a search parameter as input and extract all targeted references

Returns a dict like: {“Patient”: [“list”, “of”, “referenced”, “patient”, “ids”], “Observation”: []}

header: fhirpath.engine.base.EngineResultHeader
class fhirpath.engine.base.EngineResultBody[source]

Bases: collections.deque

add(value)[source]
append(value)[source]
class fhirpath.engine.base.EngineResultHeader(total, raw_query=None)[source]

Bases: object

elements = None
generated_on = None
raw_query = None
total = None
class fhirpath.engine.base.EngineResultRow(iterable=(), /)[source]

Bases: list

fhirpath.engine.es module

class fhirpath.engine.es.AsyncElasticsearchEngine(fhir_release, conn_factory, dialect_factory)[source]

Bases: fhirpath.engine.es.ElasticsearchEngineBase

Async Elasticsearch Engine

async execute(query, unrestricted=False, query_type=<EngineQueryType.DML: 'DML'>)[source]
classmethod is_async()[source]
async process_raw_result(rawresult, selects, query_type)[source]
class fhirpath.engine.es.ElasticsearchEngine(fhir_release, conn_factory, dialect_factory)[source]

Bases: fhirpath.engine.es.ElasticsearchEngineBase

Elasticsearch Engine

execute(query, unrestricted=False, query_type=<EngineQueryType.DML: 'DML'>)[source]
process_raw_result(rawresult, selects, query_type)[source]
class fhirpath.engine.es.ElasticsearchEngineBase(fhir_release, conn_factory, dialect_factory)[source]

Bases: fhirpath.engine.base.Engine

build_security_query(query)[source]
calculate_field_index_name(resource_type)[source]
current_url()[source]

complete url from current request return yarl.URL

extract_hits(source_filters, hits, container, doc_type='_doc')[source]
generate_mappings(reference_analyzer: Optional[str] = None, token_normalizer: Optional[str] = None)[source]

You may use this function to build the ES mapping. Returns an object like: {

“Patient”: {
“properties”: {
“identifier”: {
“properties”: {
“use”: {

“type”: “keyword”, “index”: true, “store”: false, “fields”: {

“raw”: {

“type”: “keyword”

}

}

}

}

}

}

}

}

get_index_name(resource_type: Optional[str] = None)[source]
get_mapping(resource_type)[source]
initial_bundle_data()[source]

Can be overridden in sub class

wrapped_with_bundle(result, includes=None, as_json=False)[source]
fhirpath.engine.es.navigate_indexed_path(source, path_)[source]

fhirpath.engine.fhirbase module

class fhirpath.engine.fhirbase.FhirbaseEngine(fhir_release, conn_factory, dialect_factory)[source]

Bases: fhirpath.engine.base.Engine

Module contents

Implementation, the medium result collected from ES server

class fhirpath.engine.Connection(conn)[source]

Bases: object

classmethod from_config(config: dict)[source]
classmethod from_prepared(conn)[source]

Connection instance creation, using already prepared RAW connection

classmethod from_url(url: str)[source]

1.) may be use connector utilities 2.) may be url parser

classmethod is_async()[source]
property raw_connection
class fhirpath.engine.Engine(fhir_release, conn_factory, dialect_factory)[source]

Bases: abc.ABC

Idea: # 1.) https://docs.sqlalchemy.org/en/13/core/ # connections.html#sqlalchemy.engine.Engine.connect 2.) https://docs.sqlalchemy.org/en/13/core/ connections.html#sqlalchemy.engine.Connection 3.) Dialect could have raw connection, query compiler 4.) Engine would have execute and result processing through provider, yes provider!

before_execute(query)[source]

Hook: before execution of query

create_connection(factory)[source]
create_dialect(factory)[source]
classmethod is_async()[source]
class fhirpath.engine.EngineResult(header: fhirpath.engine.base.EngineResultHeader, body: fhirpath.engine.base.EngineResultBody)[source]

Bases: object

body: fhirpath.engine.base.EngineResultBody
extract_ids() → Dict[str, List[str]][source]
extract_references(search_param: fhirpath.fhirspec.spec.SearchParameter) → Dict[str, List[str]][source]

Takes a search parameter as input and extract all targeted references

Returns a dict like: {“Patient”: [“list”, “of”, “referenced”, “patient”, “ids”], “Observation”: []}

header: fhirpath.engine.base.EngineResultHeader
class fhirpath.engine.EngineResultBody[source]

Bases: collections.deque

add(value)[source]
append(value)[source]
class fhirpath.engine.EngineResultHeader(total, raw_query=None)[source]

Bases: object

elements = None
generated_on = None
raw_query = None
total = None
class fhirpath.engine.EngineResultRow(iterable=(), /)[source]

Bases: list