RESTfull API Service¶
Server Side codex examples¶
Search aka GET Service¶
# -*- coding: utf-8 -*-
from plone import api
from plone.app.fhirfield.helpers import parse_query_string
from plone.restapi.services import Service
from zope.interface import implementer
from zope.publisher.interfaces import IPublishTraverse
@implementer(IPublishTraverse)
class FHIRSearchService(Service):
""" """
def __init__(self, context, request):
""" """
super(FHIRSearchService, self).__init__(context, request)
self.params = []
def reply(self):
""" """
results = [getattr(
r.getObject(),
'{0}_resource'.format(self.resource_type).lower()).as_json()
for r in self.build_result()]
if self.resource_id:
if not results:
self.request.response.setStatus(404)
return None
return results[0]
return results
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_params(self):
"""We are not using self.request.form (parsed by Zope Publisher)!!
There is special meaning for colon(:) in key field. For example `field_name:list`
treats data as List and it doesn't recognize FHIR search modifier like :not, :missing
as a result, from colon(:) all chars are ommited.
"""
if self.resource_id:
return {'_id': self.resource_id}
return parse_query_string(self.request)
def build_query(self):
""" """
query = dict()
fhir_query = self.get_params()
extra_params = dict()
# not supporting count yet!
if '_count' in fhir_query:
extra_params['_count'] = fhir_query.pop('_count')
if 'search-offset' in fhir_query:
extra_params['search-offset'] = fhir_query.pop('search-offset')
if 'search-id' in fhir_query:
extra_params['search-id'] = fhir_query.pop('search-id')
if fhir_query:
query[self._get_fhir_fieldname(self.resource_type)] = \
fhir_query
return query, extra_params
def build_result(self):
""" """
query, extra_params = self.build_query()
results = api.content.find(**query) # noqa: P001
return results
FHIR Resource Add aka POST Service¶
# -*- coding: utf-8 -*-
from Acquisition import aq_base
from Acquisition.interfaces import IAcquirer
from plone import api
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(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 'error' in response:
self.request.response.setStatus(400)
return response
def _create_object(self, fhir):
""" """
form_data = {
'@type': 'FF' + 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 = api.portal.get()
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).as_json()
self.request.response.setHeader(
'Location',
'/'.join([self.context.portal_url(),
'@fhir',
response['resourceType'],
response['id']]))
return response
FHIR Resource Update aka PATCH Service¶
# -*- coding: utf-8 -*-
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.interface import alsoProvides
from zope.interface import implementer
from zope.publisher.interfaces import IPublishTraverse
import plone.protect.interfaces
@implementer(IPublishTraverse)
class FHIRResourcePatch(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
@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 = {
'{0}_resource'.format(
self.resource_type.lower()
): {'_id': self.resource_id}
}
brains = api.content.find(**query)
if len(brains) == 0:
self.request.response.setStatus(404)
return None
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
return None
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-31&_lastUpdated=gt2017-01-01')
>>> response.status_code
200
>>> len(response.json())
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')
>>> len(response.json())
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