jimfuqian/BB2-3679-Expose-PACA-data-Claim-and-ClaimResponse-thru-BB2-API

apps/accounts/fixtures/scopes.json

@@ -80,5 +80,29 @@
             "protected_resources": "[[\"POST\", \"/v[12]/o/introspect\"]]",
             "default": "False"
         }
+    },
+    {
+        "model": "capabilities.protectedcapability",
+        "pk": 7,
+        "fields": {
+            "title": "My Medicare partially adjudicated claims.",
+            "slug": "patient/Claim.read",
+            "group": 5,
+            "description": "Claim FHIR Resource",
+            "protected_resources": "[\n  [\n    \"POST\",\n   \"/v2/fhir/Claim/_search$\"\n  ],\n  [\n    \"GET\",\n   \"/v2/fhir/Claim/(?P<resource_id>[^/]+)$\"\n  ]\n]",
+            "default": "True"
+        }
+    },
+    {
+        "model": "capabilities.protectedcapability",
+        "pk": 8,
+        "fields": {
+            "title": "My Medicare partially adjudicated claim responses.",
+            "slug": "patient/ClaimResponse.read",
+            "group": 5,
+            "description": "ClaimResponse FHIR Resource",
+            "protected_resources": "[\n  [\n    \"POST\",\n   \"/v2/fhir/ClaimResponse/_search$\"\n  ],\n  [\n    \"GET\",\n   \"/v2/fhir/ClaimResponse/(?P<resource_id>[^/]+)$\"\n  ]\n]",
+            "default": "True"
+        }
     }
 ]

apps/authorization/permissions.py

@@ -32,10 +32,10 @@ def has_object_permission(self, request, view, obj):
         # Patient resources were taken care of above
         # Return 404 on error to avoid notifying unauthorized user the object exists
 
-        return is_resource_for_patient(obj, request.crosswalk.fhir_id)
+        return is_resource_for_patient(obj, request.crosswalk.fhir_id, request.crosswalk.user_mbi)
 
 
-def is_resource_for_patient(obj, patient_id):
+def is_resource_for_patient(obj, patient_id, user_mbi):
     try:
         if obj['resourceType'] == 'Coverage':
             reference = obj['beneficiary']['reference']
@@ -51,9 +51,15 @@ def is_resource_for_patient(obj, patient_id):
             reference_id = obj['id']
             if reference_id != patient_id:
                 raise exceptions.NotFound()
+        elif obj['resourceType'] == 'Claim':
+            if not _check_mbi(obj, user_mbi):
+                raise exceptions.NotFound()
+        elif obj['resourceType'] == 'ClaimResponse':
+            if not _check_mbi(obj, user_mbi):
+                raise exceptions.NotFound()
         elif obj['resourceType'] == 'Bundle':
             for entry in obj.get('entry', []):
-                is_resource_for_patient(entry['resource'], patient_id)
+                is_resource_for_patient(entry['resource'], patient_id, user_mbi)
         else:
             raise exceptions.NotFound()
 
@@ -62,3 +68,22 @@ def is_resource_for_patient(obj, patient_id):
     except Exception:
         return False
     return True
+
+
+# helper verify mbi of a claim or claim response resource
+def _check_mbi(obj, mbi):
+    matched = False
+    try:
+        if obj['contained']:
+            for c in obj['contained']:
+                if c['resourceType'] == 'Patient':
+                    identifiers = c['identifier']
+                    if len(identifiers) > 0:
+                        if identifiers[0]['value'] == mbi:
+                            matched = True
+                            break
+    except KeyError as ke:
+        # log error and return false
+        print(ke)
+        pass
+    return matched

apps/capabilities/management/commands/create_blue_button_scopes.py

@@ -98,6 +98,42 @@ def create_coverage_capability(group,
     return c
 
 
+def create_claim_capability(group,
+                               fhir_prefix,
+                               title="My Medicare partially adjudicated claims."):
+    c = None
+    description = "Claim FHIR Resource"
+    smart_scope_string = "patient/Claim.read"
+    pr = []
+    pr.append(["GET", "%sClaim/" % fhir_prefix])
+    pr.append(["GET", "%sClaim/[id]" % fhir_prefix])
+    if not ProtectedCapability.objects.filter(slug=smart_scope_string).exists():
+        c = ProtectedCapability.objects.create(group=group,
+                                               title=title,
+                                               description=description,
+                                               slug=smart_scope_string,
+                                               protected_resources=json.dumps(pr, indent=4))
+    return c
+
+
+def create_claimresponse_capability(group,
+                               fhir_prefix,
+                               title="My Medicare partially adjudicated claim responses."):
+    c = None
+    description = "ClaimResponse FHIR Resource"
+    smart_scope_string = "patient/ClaimResponse.read"
+    pr = []
+    pr.append(["GET", "%sClaimResponse/" % fhir_prefix])
+    pr.append(["GET", "%sClaimResponse/[id]" % fhir_prefix])
+    if not ProtectedCapability.objects.filter(slug=smart_scope_string).exists():
+        c = ProtectedCapability.objects.create(group=group,
+                                               title=title,
+                                               description=description,
+                                               slug=smart_scope_string,
+                                               protected_resources=json.dumps(pr, indent=4))
+    return c
+
+
 class Command(BaseCommand):
     help = 'Create BlueButton Group and Scopes'
 

apps/dot_ext/migrations/0009_internalapplicationlabelsproxy_and_more.py

@@ -0,0 +1,31 @@
+# Generated by Django 4.2.17 on 2025-02-19 23:35
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('dot_ext', '0008_internalapplicationlabels_and_more'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='InternalApplicationLabelsProxy',
+            fields=[
+            ],
+            options={
+                'verbose_name': 'Internal Category',
+                'verbose_name_plural': 'Internal Categories',
+                'proxy': True,
+                'indexes': [],
+                'constraints': [],
+            },
+            bases=('dot_ext.internalapplicationlabels',),
+        ),
+        migrations.AlterField(
+            model_name='application',
+            name='internal_application_labels',
+            field=models.ManyToManyField(blank=True, to='dot_ext.internalapplicationlabels'),
+        ),
+    ]

apps/fhir/bluebutton/constants.py

@@ -1,3 +1,3 @@
-ALLOWED_RESOURCE_TYPES = ['Patient', 'Coverage', 'ExplanationOfBenefit']
+ALLOWED_RESOURCE_TYPES = ['Patient', 'Coverage', 'ExplanationOfBenefit', 'Claim', 'ClaimResponse']
 DEFAULT_PAGE_SIZE = 10
 MAX_PAGE_SIZE = 50

apps/fhir/bluebutton/migrations/0005_crosswalk__user_mbi.py

@@ -0,0 +1,18 @@
+# Generated by Django 4.2.17 on 2025-02-19 23:35
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('bluebutton', '0004_createnewapplication_mycredentialingrequest'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='crosswalk',
+            name='_user_mbi',
+            field=models.CharField(db_column='user_mbi', db_index=True, default=None, max_length=32, null=True, unique=True, verbose_name='User MBI ID'),
+        ),
+    ]

apps/fhir/bluebutton/models.py

@@ -120,6 +120,18 @@ class Crosswalk(models.Model):
         db_index=True,
     )
 
+    # This stores the MBI value.
+    #     Can be null for backwards migration compatibility.
+    _user_mbi = models.CharField(
+        max_length=32,
+        verbose_name="User MBI ID",
+        unique=True,
+        null=True,
+        default=None,
+        db_column="user_mbi",
+        db_index=True,
+    )
+
     objects = models.Manager()  # Default manager
     real_objects = RealCrosswalkManager()  # Real bene manager
     synth_objects = SynthCrosswalkManager()  # Synth bene manager
@@ -145,6 +157,10 @@ def user_hicn_hash(self):
     def user_mbi_hash(self):
         return self._user_mbi_hash
 
+    @property
+    def user_mbi(self):
+        return self._user_mbi
+
     @user_hicn_hash.setter
     def user_hicn_hash(self, value):
         self._user_id_hash = value
@@ -153,6 +169,10 @@ def user_hicn_hash(self, value):
     def user_mbi_hash(self, value):
         self._user_mbi_hash = value
 
+    @user_mbi.setter
+    def user_mbi(self, value):
+        self._user_mbi = value
+
 
 class ArchivedCrosswalk(models.Model):
     """

apps/fhir/bluebutton/permissions.py

@@ -52,6 +52,12 @@ def has_object_permission(self, request, view, obj):
                 reference_id = reference.split("/")[1]
                 if reference_id != request.crosswalk.fhir_id:
                     raise exceptions.NotFound()
+            elif request.resource_type == "Claim":
+                if not _check_mbi(obj, request.crosswalk.user_mbi):
+                    raise exceptions.NotFound()
+            elif request.resource_type == "ClaimResponse":
+                if not _check_mbi(obj, request.crosswalk.user_mbi):
+                    raise exceptions.NotFound()
             else:
                 reference_id = obj["id"]
                 if reference_id != request.crosswalk.fhir_id:
@@ -68,7 +74,6 @@ def has_object_permission(self, request, view, obj):
 class SearchCrosswalkPermission(HasCrosswalk):
     def has_object_permission(self, request, view, obj):
         patient_id = request.crosswalk.fhir_id
-
         if "patient" in request.GET and request.GET["patient"] != patient_id:
             return False
 
@@ -98,3 +103,22 @@ def has_permission(self, request, view):
             )
 
         return True
+
+
+# helper verify mbi of a claim or claim response resource
+def _check_mbi(obj, mbi):
+    matched = False
+    try:
+        if obj['contained']:
+            for c in obj['contained']:
+                if c['resourceType'] == 'Patient':
+                    identifiers = c['identifier']
+                    if len(identifiers) > 0:
+                        if identifiers[0]['value'] == mbi:
+                            matched = True
+                            break
+    except KeyError as ke:
+        # log error and return false
+        print(ke)
+        pass
+    return matched

apps/fhir/bluebutton/v2/urls.py

@@ -5,11 +5,15 @@
     ReadViewCoverage,
     ReadViewExplanationOfBenefit,
     ReadViewPatient,
+    ReadViewClaim,
+    ReadViewClaimResponse,
 )
 from apps.fhir.bluebutton.views.search import (
     SearchViewCoverage,
     SearchViewExplanationOfBenefit,
     SearchViewPatient,
+    SearchViewClaim,
+    SearchViewClaimResponse,
 )
 
 admin.autodiscover()
@@ -51,4 +55,33 @@
         SearchViewExplanationOfBenefit.as_view(version=2),
         name="bb_oauth_fhir_eob_search_v2",
     ),
+    # Claim SearchView
+    re_path(
+        r"Claim/_search$",
+        SearchViewClaim.as_view(version=2),
+        name="bb_oauth_fhir_claim_search",
+    ),
+    re_path(
+        r"ClaimJSON/_search$",
+        SearchViewClaim.as_view(version=2),
+        name="bb_oauth_fhir_claimjson_search",
+    ),
+    # Claim ReadView
+    re_path(
+        r"Claim/(?P<resource_id>[^/]+)",
+        ReadViewClaim.as_view(version=2),
+        name="bb_oauth_fhir_claim_read",
+    ),
+    # ClaimResponse SearchView
+    re_path(
+        r"ClaimResponse/_search$",
+        SearchViewClaimResponse.as_view(version=2),
+        name="bb_oauth_fhir_claimresponse_search",
+    ),
+    # ClaimResponse ReadView
+    re_path(
+        r"ClaimResponse/(?P<resource_id>[^/]+)",
+        ReadViewClaimResponse.as_view(version=2),
+        name="bb_oauth_fhir_claimresponse_read",
+    ),
 ]