refactor the fhirapi views and util functions

Author: clewellyn-nava

apps/fhir/bluebutton/permissions.py

@@ -72,7 +72,7 @@ def has_object_permission(self, request, view, obj):
 
 
 class SearchCrosswalkPermission(HasCrosswalk):
-    def has_object_permission(self, request, view, obj):
+    def has_object_permission(self, request, view, obj):  # type: ignore
         if view.version in Versions.supported_versions():
             patient_id = request.crosswalk.fhir_id(view.version)
         else:

apps/fhir/bluebutton/tests/test_utils.py

@@ -7,6 +7,8 @@
 from apps.test import BaseApiTest
 from apps.fhir.bluebutton.models import Crosswalk
 from apps.versions import Versions
+from apps.fhir.server.settings import fhir_settings
+
 
 from apps.fhir.bluebutton.utils import (
     notNone,
@@ -16,7 +18,6 @@
     prepend_q,
     dt_patient_reference,
     crosswalk_patient_id,
-    get_resourcerouter,
     build_oauth_resource,
     valid_patient_read_or_search_call,
 )
@@ -123,13 +124,12 @@ def test_FhirServerAuth(self):
 
         """ Test 1: pass nothing"""
 
-        resource_router = get_resourcerouter()
         expected = {}
-        expected['client_auth'] = resource_router.client_auth
+        expected['client_auth'] = fhir_settings.client_auth
         expected['cert_file'] = os.path.join(settings.FHIR_CLIENT_CERTSTORE,
-                                             resource_router.cert_file)
+                                             fhir_settings.cert_file)
         expected['key_file'] = os.path.join(settings.FHIR_CLIENT_CERTSTORE,
-                                            resource_router.key_file)
+                                            fhir_settings.key_file)
 
         response = FhirServerAuth()
 

apps/fhir/bluebutton/urls.py

@@ -2,14 +2,14 @@
 from django.contrib import admin
 
 from apps.fhir.bluebutton.views.read import (
+    ReadViewPatient,
     ReadViewCoverage,
     ReadViewExplanationOfBenefit,
-    ReadViewPatient,
 )
 from apps.fhir.bluebutton.views.search import (
+    SearchViewPatient,
     SearchViewCoverage,
     SearchViewExplanationOfBenefit,
-    SearchViewPatient,
 )
 
 admin.autodiscover()

apps/fhir/bluebutton/utils.py

@@ -221,20 +221,15 @@ def request_call(request, call_url, crosswalk=None, timeout=None, get_parameters
     call_url = target server URL and search parameters to be sent
     crosswalk = Crosswalk record. The crosswalk is keyed off Request.user
     timeout allows a timeout in seconds to be set.
-
-    FhirServer is joined to Crosswalk.
-    FhirServerAuth and FhirServerVerify receive crosswalk and lookup
-       values in the linked fhir_server model.
-
     """
 
     logger_perf = bb2logging.getLogger(bb2logging.PERFORMANCE_LOGGER, request)
 
     # Updated to receive crosswalk (Crosswalk entry for user)
     # call FhirServer_Auth(crosswalk) to get authentication
-    auth_state = FhirServerAuth(crosswalk)
+    auth_state = FhirServerAuth()
 
-    verify_state = FhirServerVerify(crosswalk)
+    verify_state = fhir_settings.verify_server
     if auth_state["client_auth"]:
         # cert puts cert and key file together
         # (cert_file_path, key_file_path)
@@ -326,16 +321,21 @@ def notNone(value=None, default=None):
         return value
 
 
-def FhirServerAuth(crosswalk=None):
-    # Get default clientauth settings from base.py
-    # Receive a crosswalk.id or None
-    # Return a dict
+def FhirServerAuth() -> dict:
+    """Helper class to modify cert paths if client_auth is true
+    TODO - this can probably be refactored or removed, rolled into the FHIRServerSettings class
+
+    Returns:
+        dict: A dictionary with the following:
+            - client_auth (bool): if client authentication is required
+            - cert_file (str): path to cert file
+            - key_file (str): path to key file
+    """
 
     auth_settings = {}
-    resource_router = get_resourcerouter()
-    auth_settings["client_auth"] = resource_router.client_auth
-    auth_settings["cert_file"] = resource_router.cert_file
-    auth_settings["key_file"] = resource_router.key_file
+    auth_settings["client_auth"] = fhir_settings.client_auth
+    auth_settings["cert_file"] = fhir_settings.cert_file
+    auth_settings["key_file"] = fhir_settings.key_file
 
     if auth_settings["client_auth"]:
         # join settings.FHIR_CLIENT_CERTSTORE to cert_file and key_file
@@ -351,12 +351,6 @@ def FhirServerAuth(crosswalk=None):
     return auth_settings
 
 
-def FhirServerVerify(crosswalk=None):
-    # Get default Server Verify Setting
-    # Return True or False (Default)
-    return get_resourcerouter().verify_server
-
-
 def mask_with_this_url(request, host_path="", in_text="", find_url=""):
     """find_url in in_text and replace with url for this server"""
 
@@ -455,10 +449,6 @@ def get_crosswalk(user):
     return None
 
 
-def get_resourcerouter(crosswalk=None):
-    return fhir_settings
-
-
 def handle_http_error(e):
     """Handle http error from request_call
 
@@ -605,7 +595,7 @@ def get_response_text(fhir_response=None):
         return text_in
 
 
-def build_oauth_resource(request, version=Versions.NOT_AN_API_VERSION, format_type="json"):
+def build_oauth_resource(request, version=Versions.NOT_AN_API_VERSION, format_type="json") -> dict | str:
     """
     Create a resource entry for oauth endpoint(s) for insertion
     into the conformance/capabilityStatement
@@ -707,17 +697,15 @@ def get_v2_patient_by_id(id, request):
     a helper adapted to just get patient given an id out of band of auth flow
     or normal data flow, use by tools such as BB2-Tools admin viewers
     """
-    auth_settings = FhirServerAuth(None)
+    auth_settings = FhirServerAuth()
     certs = (auth_settings["cert_file"], auth_settings["key_file"])
 
     headers = generate_info_headers(request)
     headers["BlueButton-Application"] = "BB2-Tools"
     headers["includeIdentifiers"] = "true"
     # for now this will only work for v1/v2 patients, but we'll need to be able to
     # determine if the user is V3 and use those endpoints later
-    url = "{}/v2/fhir/Patient/{}?_format={}".format(
-        get_resourcerouter().fhir_url, id, settings.FHIR_PARAM_FORMAT
-    )
+    url = f'{fhir_settings.fhir_url}/v2/fhir/Patient/{id}?_format={settings.FHIR_PARAM_FORMAT}'
     s = requests.Session()
     req = requests.Request("GET", url, headers=headers)
     prepped = req.prepare()
@@ -730,17 +718,15 @@ def get_v2_patient_by_id(id, request):
 # of the ticket to remove the user_mbi_hash column from the crosswalk table
 # We can remove this entire function at that point
 def get_patient_by_mbi_hash(mbi_hash, request):
-    auth_settings = FhirServerAuth(None)
+    auth_settings = FhirServerAuth()
     certs = (auth_settings["cert_file"], auth_settings["key_file"])
     headers = generate_info_headers(request)
     headers["BlueButton-Application"] = "BB2-Tools"
     headers["includeIdentifiers"] = "true"
 
     search_identifier = f'https://bluebutton.cms.gov/resources/identifier/mbi-hash|{mbi_hash}'  # noqa: E231
     payload = {'identifier': search_identifier}
-    url = '{}/v2/fhir/Patient/_search'.format(
-        get_resourcerouter().fhir_url
-    )
+    url = f'{fhir_settings.fhir_url}/v2/fhir/Patient/_search'
 
     s = requests.Session()
     req = requests.Request("POST", url, headers=headers, data=payload)

apps/fhir/bluebutton/v2/urls.py

@@ -4,28 +4,27 @@
 from apps.fhir.bluebutton.views.read import (
     ReadViewCoverage,
     ReadViewExplanationOfBenefit,
-    ReadViewPatient,
 )
 from apps.fhir.bluebutton.views.search import (
     SearchViewCoverage,
     SearchViewExplanationOfBenefit,
-    SearchViewPatient,
 )
+from apps.fhir.bluebutton.views.patient_viewset import PatientViewSet
 
 admin.autodiscover()
 
 urlpatterns = [
     # Patient ReadView
     re_path(
-        r"Patient/(?P<resource_id>[^/]+)",
-        ReadViewPatient.as_view(version=2),
-        name="bb_oauth_fhir_patient_read_or_update_or_delete_v2",
+        r'Patient/(?P<resource_id>[^/]+)',
+        PatientViewSet.as_view({'get': 'retrieve'}, version=2),
+        name='bb_oauth_fhir_patient_read_or_update_or_delete_v2',
     ),
     # Patient SearchView
     re_path(
-        r"Patient[/]?",
-        SearchViewPatient.as_view(version=2),
-        name="bb_oauth_fhir_patient_search_v2",
+        r'Patient[/]?',
+        PatientViewSet.as_view({'get': 'list'}, version=2),
+        name='bb_oauth_fhir_patient_search_v2',
     ),
     # Coverage ReadView
     re_path(

apps/fhir/bluebutton/v3/urls.py

@@ -5,56 +5,58 @@
 from apps.fhir.bluebutton.views.read import (
     ReadViewCoverage,
     ReadViewExplanationOfBenefit,
-    ReadViewPatient,
 )
 from apps.fhir.bluebutton.views.search import (
     SearchViewCoverage,
     SearchViewExplanationOfBenefit,
-    SearchViewPatient,
 )
+from apps.fhir.bluebutton.views.patient_viewset import PatientViewSet
+from apps.fhir.bluebutton.views.insurancecard_viewset import DigitalInsuranceCardViewSet
 
 admin.autodiscover()
 
 urlpatterns = [
     # Patient ReadView
     re_path(
-        r"Patient/(?P<resource_id>[^/]+)",
-        waffle_switch("v3_endpoints")(ReadViewPatient.as_view(version=3)),
-        name="bb_oauth_fhir_patient_read_or_update_or_delete_v3",
+        r'Patient/(?P<resource_id>[^/]+)',
+        waffle_switch('v3_endpoints')(PatientViewSet.as_view({'get': 'retrieve'}, version=3)),
+        name='bb_oauth_fhir_patient_read_or_update_or_delete_v3',
     ),
     # Patient SearchView
     re_path(
-        r"Patient[/]?",
-        waffle_switch("v3_endpoints")(SearchViewPatient.as_view(version=3)),
-        name="bb_oauth_fhir_patient_search_v3",
+        r'Patient[/]?',
+        waffle_switch('v3_endpoints')(PatientViewSet.as_view({'get': 'list'}, version=3)),
+        name='bb_oauth_fhir_patient_search_v3',
     ),
     # Coverage ReadView
     re_path(
-        r"Coverage/(?P<resource_id>[^/]+)",
-        waffle_switch("v3_endpoints")(ReadViewCoverage.as_view(version=3)),
-        name="bb_oauth_fhir_coverage_read_or_update_or_delete_v3",
+        r'Coverage/(?P<resource_id>[^/]+)',
+        waffle_switch('v3_endpoints')(ReadViewCoverage.as_view(version=3)),
+        name='bb_oauth_fhir_coverage_read_or_update_or_delete_v3',
     ),
     # Coverage SearchView
     re_path(
-        r"Coverage[/]?",
-        waffle_switch("v3_endpoints")(SearchViewCoverage.as_view(version=3)),
-        name="bb_oauth_fhir_coverage_search_v3",
+        r'Coverage[/]?',
+        waffle_switch('v3_endpoints')(SearchViewCoverage.as_view(version=3)),
+        name='bb_oauth_fhir_coverage_search_v3',
     ),
     # EOB ReadView
     re_path(
-        r"ExplanationOfBenefit/(?P<resource_id>[^/]+)",
-        waffle_switch("v3_endpoints")(ReadViewExplanationOfBenefit.as_view(version=3)),
-        name="bb_oauth_fhir_eob_read_or_update_or_delete_v3",
+        r'ExplanationOfBenefit/(?P<resource_id>[^/]+)',
+        waffle_switch('v3_endpoints')(ReadViewExplanationOfBenefit.as_view(version=3)),
+        name='bb_oauth_fhir_eob_read_or_update_or_delete_v3',
     ),
     # EOB SearchView
     re_path(
-        r"ExplanationOfBenefit[/]?",
-        waffle_switch("v3_endpoints")(SearchViewExplanationOfBenefit.as_view(version=3)),
-        name="bb_oauth_fhir_eob_search_v3",
+        r'ExplanationOfBenefit[/]?',
+        waffle_switch('v3_endpoints')(SearchViewExplanationOfBenefit.as_view(version=3)),
+        name='bb_oauth_fhir_eob_search_v3',
     ),
     # C4DIC
-    # Digital Insurance Card ReadView
+    # Digital Insurance Card ViewSet
     re_path(
-        r'', waffle_switch
-    )
+        r'DigitalInsuranceCard[/]?',
+        waffle_switch('v3_endpoints')(DigitalInsuranceCardViewSet.as_view({'get': 'list'}, version=3)),
+        name='bb_oauth_fhir_dic_read',
+    ),
 ]