RESTfull API Service¶
Server Side codex examples¶
Search aka GET Service¶
# -*- coding: utf-8 -*-
from collective.elasticsearch.es import ElasticSearchCatalog
from collective.fhirpath.utils import FHIRModelServiceMixin
from fhirpath.enums import FHIR_VERSION
from fhirpath.interfaces import IElasticsearchEngineFactory
from fhirpath.interfaces import IFhirSearch
from fhirpath.interfaces import ISearchContextFactory
from plone import api
from plone.restapi.services import Service
from zope.component import queryMultiAdapter
from zope.interface import implementer
from zope.publisher.interfaces import IPublishTraverse
@implementer(IPublishTraverse)
class FHIRSearchService(FHIRModelServiceMixin, Service):
""" """
def __init__(self, context, request):
""" """
super(FHIRSearchService, self).__init__(context, request)
self.params = []
def get_es_catalog(self):
""" """
return ElasticSearchCatalog(api.portal.get_tool("portal_catalog"))
def get_factory(self, resource_type, unrestricted=False):
""" """
factory = queryMultiAdapter(
(self.get_es_catalog(),), IElasticsearchEngineFactory
)
engine = factory(fhir_release=FHIR_VERSION.STU3)
context = queryMultiAdapter((engine,), ISearchContextFactory)(
resource_type, unrestricted=unrestricted
)
factory = queryMultiAdapter((context,), IFhirSearch)
return factory
def reply(self):
""" """
bundle = self.build_result()
if self.resource_id:
if bundle.total == 0:
return self.reply_no_content(404)
return bundle.entry[0].resource
return bundle
def publishTraverse(self, request, name): # noqa: N802
# Consume any path segments after /@fhir as parameters
self.params.append(name)
return self
@property
def resource_id(self):
""" """
if 1 < len(self.params):
return self.params[1]
return None
@property
def resource_type(self):
""" """
if 0 < len(self.params):
_rt = self.params[0]
return _rt
return None
def _get_fhir_fieldname(self, resource_type=None):
"""We assume FHIR Field name is ``{resource type}_resource``"""
resource_type = resource_type or self.resource_type
return "{0}_resource".format(resource_type.lower())
def get_query_string(self):
""" """
if self.resource_id:
return "_id={0}".format(self.resource_id)
return self.request["QUERY_STRING"]
def build_result(self):
""" """
factory = self.get_factory(self.resource_type)
return factory(query_string=self.get_query_string())
FHIR Resource Add aka POST Service¶
# -*- coding: utf-8 -*-
from Acquisition import aq_base
from Acquisition.interfaces import IAcquirer
from collective.fhirpath.utils import FHIRModelServiceMixin
from plone.restapi.deserializer import json_body
from plone.restapi.exceptions import DeserializationError
from plone.restapi.interfaces import IDeserializeFromJson
from plone.restapi.services import Service
from plone.restapi.services.content.utils import add as add_obj
from plone.restapi.services.content.utils import create as create_obj
from Products.CMFPlone.utils import safe_hasattr
from zope.component import queryMultiAdapter
from zope.event import notify
from zope.interface import alsoProvides
from zope.interface import implementer
from zope.lifecycleevent import ObjectCreatedEvent
from zope.publisher.interfaces import IPublishTraverse
import json
import plone.protect.interfaces
__author__ = "Md Nazrul Islam <nazrul@zitelab.dk>"
@implementer(IPublishTraverse)
class FHIRResourceAdd(FHIRModelServiceMixin, Service):
"""Creates a new FHIR Resource object.
"""
def __init__(self, context, request):
""" """
super(FHIRResourceAdd, self).__init__(context, request)
self.params = []
def publishTraverse(self, request, name): # noqa: N802
# Consume any path segments after /@fhir as parameters
self.params.append(name)
return self
@property
def resource_type(self):
""" """
if 0 < len(self.params):
_rt = self.params[0]
return _rt
return None
def reply(self):
""" """
data = json_body(self.request)
# Disable CSRF protection
if "IDisableCSRFProtection" in dir(plone.protect.interfaces):
alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection)
response = self._create_object(data)
if isinstance(response, dict) and "error" in response:
self.request.response.setStatus(400)
return response
def _create_object(self, fhir):
""" """
form_data = {
"@type": fhir["resourceType"],
"id": fhir["id"],
"title": "{0}-{1}".format(self.resource_type, fhir["id"]),
}
fhir_field_name = "{0}_resource".format(fhir["resourceType"].lower())
form_data[fhir_field_name] = fhir
self.request["BODY"] = json.dumps(form_data)
context = self.context
obj = create_obj(
context, form_data["@type"], id_=form_data["id"], title=form_data["title"]
)
if isinstance(obj, dict) and "error" in obj:
self.request.response.setStatus(400)
return obj
# Acquisition wrap temporarily to satisfy things like vocabularies
# depending on tools
temporarily_wrapped = False
if IAcquirer.providedBy(obj) and not safe_hasattr(obj, "aq_base"):
obj = obj.__of__(context)
temporarily_wrapped = True
# Update fields
deserializer = queryMultiAdapter((obj, self.request), IDeserializeFromJson)
if deserializer is None:
self.request.response.setStatus(501)
return dict(
error=dict(
message="Cannot deserialize type {0}".format(obj.portal_type)
)
)
try:
deserializer(validate_all=True, create=True)
except DeserializationError as e:
self.request.response.setStatus(400)
return dict(error=dict(type="DeserializationError", message=str(e)))
if temporarily_wrapped:
obj = aq_base(obj)
# Notify Dexterity Created
if not getattr(deserializer, "notifies_create", False):
notify(ObjectCreatedEvent(obj))
# Adding to Container
add_obj(context, obj, rename=False)
self.request.response.setStatus(201)
response = getattr(obj, fhir_field_name)
self.request.response.setHeader(
"Location",
"/".join(
[
self.context.portal_url(),
"@fhir",
response.resource_type,
response.id,
]
),
)
return response
FHIR Resource Update aka PATCH Service¶
# -*- coding: utf-8 -*-
from collective.elasticsearch.es import ElasticSearchCatalog
from collective.fhirpath.interfaces import IZCatalogFhirSearch
from collective.fhirpath.utils import FHIRModelServiceMixin
from fhirpath.enums import FHIR_VERSION
from fhirpath.interfaces import IElasticsearchEngineFactory
from fhirpath.interfaces import ISearchContextFactory
from plone import api
from plone.restapi.deserializer import json_body
from plone.restapi.services import Service
from plone.restapi.services.locking.locking import is_locked
from zope.component import queryMultiAdapter
from zope.interface import alsoProvides
from zope.interface import implementer
from zope.publisher.interfaces import IPublishTraverse
import plone.protect.interfaces
@implementer(IPublishTraverse)
class FHIRResourcePatch(FHIRModelServiceMixin, Service):
"""Patch a FHIR Resource object.
"""
def __init__(self, context, request):
""" """
super(FHIRResourcePatch, self).__init__(context, request)
self.params = []
def publishTraverse(self, request, name): # noqa: N802
# Consume any path segments after /@fhir as parameters
self.params.append(name)
return self
def get_es_catalog(self):
""" """
return ElasticSearchCatalog(api.portal.get_tool("portal_catalog"))
def get_factory(self, resource_type, unrestricted=False):
""" """
factory = queryMultiAdapter(
(self.get_es_catalog(),), IElasticsearchEngineFactory
)
engine = factory(fhir_release=FHIR_VERSION.STU3)
context = queryMultiAdapter((engine,), ISearchContextFactory)(
resource_type, unrestricted=unrestricted
)
factory = queryMultiAdapter((context,), IZCatalogFhirSearch)
return factory
@property
def resource_id(self):
""" """
if 1 < len(self.params):
return self.params[1]
return None
@property
def resource_type(self):
""" """
if 0 < len(self.params):
_rt = self.params[0]
return _rt
return None
def reply(self):
""" """
query_string = "_id={0}".format(self.resource_id)
factory = self.get_factory(self.resource_type)
brains = factory(query_string=query_string)
if len(brains) == 0:
self.reply_no_content(404)
obj = brains[0].getObject()
if is_locked(obj, self.request):
self.request.response.setStatus(403)
return dict(error=dict(type="Forbidden", message="Resource is locked."))
data = json_body(self.request)
# Disable CSRF protection
if "IDisableCSRFProtection" in dir(plone.protect.interfaces):
alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection)
fhir_value = getattr(obj, "{0}_resource".format(self.resource_type.lower()))
fhir_value.patch(data["patch"])
self.request.response.setStatus(204)
# Return None
self.reply_no_content(204)
REST Service registration (configuration.zcml)¶
<configure xmlns="http://namespaces.zope.org/zope"
xmlns:plone="http://namespaces.plone.org/plone"
xmlns:zcml="http://namespaces.zope.org/zcml">
<include package="plone.rest" file="configure.zcml" />
<plone:service method="GET"
name="@fhir"
for="Products.CMFCore.interfaces.ISiteRoot"
factory=".get.FHIRSearchService"
permission="zope2.View"
/>
<plone:service method="POST" name="@fhir" for="Products.CMFCore.interfaces.ISiteRoot" factory=".post.FHIRResourceAdd" permission="cmf.ManagePortal" />
<plone:service method="PATCH" name="@fhir" for="Products.CMFCore.interfaces.ISiteRoot" factory=".patch.FHIRResourcePatch" permission="cmf.ManagePortal" />
</configure>
REST Client Examples¶
Getting single resource, here we are getting Patient resource by ID.
Example(1):
>>> response = admin_session.get('/@fhir/Patient/19c5245f-89a8-49f8-b244-666b32adb92e')
>>> response.status_code
200
>>> response.json()['resourceType'] == 'Patient'
True
>>> response = admin_session.get('/@fhir/Patient/19c5245f-fake-id')
>>> response.status_code
404
Search Observation by Patient reference with status condition. Any observation until December 2017 and earlier than January 2017.
Example(2):
>>> response = admin_session.get('/@fhir/Observation?patient=Patient/19c5245f-89a8-49f8-b244-666b32adb92e&status=final&_lastUpdated=lt2017-12-31T00%3A00%3A00%2B00%3A00&_lastUpdated=gt2017-01-01T00%3A00%3A00%2B00%3A00')
>>> response.status_code
200
>>> response.json()["total"]
1
Add FHIR Resource through REST API
Example(3):
>>> import os
>>> import json
>>> import uuid
>>> import DateTime
>>> import time
>>> with open(os.path.join(FIXTURE_PATH, 'Patient.json'), 'r') as f:
... fhir_json = json.load(f)
>>> fhir_json['id'] = str(uuid.uuid4())
>>> fhir_json['name'][0]['text'] = 'Another Patient'
>>> response = admin_session.post('/@fhir/Patient', json=fhir_json)
>>> response.status_code
201
>>> time.sleep(1)
>>> response = admin_session.get('/@fhir/Patient?active=true')
>>> response.json()["total"]
2
Update (PATCH) FHIR Resource the Patient is currently activated, we will deactive.
Example(4):
>>> patch = [{'op': 'replace', 'path': '/active', 'value': False}]
>>> response = admin_session.patch('/@fhir/Patient/19c5245f-89a8-49f8-b244-666b32adb92e', json={'patch': patch})
>>> response.status_code
204