Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 90 additions & 0 deletions gateway-api/src/gateway_api/provider_request.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
"""
Module: gateway_api.provider_request

This module contains the GPProvider class, which provides a
simple client for GPProvider FHIR GP System.
The GPProvider class has a sigle method to get_structure_record which
can be used to fetch patient records from a GPProvider FHIR API endpoint.
Usage:

instantiate a GPProvider with:
provider_endpoint
provider_ASID
consumer_ASID

method get_structured_record with (may add optional parameters later):
Parameters: parameters resource

returns the response from the provider FHIR API.

"""

# imports
import requests
from requests import Response


# definitions
class ExternalServiceError(Exception):
"""
Raised when the downstream PDS request fails.

This module catches :class:`requests.HTTPError` thrown by
``response.raise_for_status()`` and re-raises it as ``ExternalServiceError`` so
callers are not coupled to ``requests`` exception types.
"""


class GPProviderClient:
"""
A simple client for GPProvider FHIR GP System.
"""

def __init__(
self,
provider_endpoint: str,
provider_asid: str,
consumer_asid: str,
) -> None:
"""
Create a GPProviderClient instance.

Args:
provider_endpoint (str): The FHIR API endpoint for the provider.
provider_asid (str): The ASID for the provider.
consumer_asid (str): The ASID for the consumer.

methods:
access_structured_record: fetch structured patient record
from GPProvider FHIR API.
"""
self.provider_endpoint = provider_endpoint
self.provider_asid = provider_asid
self.consumer_asid = consumer_asid

def access_structured_record(self) -> Response:
"""
Fetch a structured patient record from the GPProvider FHIR API.

Args:
parameters (dict): The parameters resource to send in the request.
returns:
dict: The response from the GPProvider FHIR API.
"""
response = requests.post(
self.provider_endpoint,
headers={
"Ssp-To": self.provider_asid, # alias here to match GP connect header
"Ssp-From": self.consumer_asid, # alias here to match GP connect header
},
timeout=None, # noqa: S113 quicker dev cycle; adjust as needed
)

try:
response.raise_for_status()
except requests.HTTPError as err:
raise ExternalServiceError(
f"GPProvider FHIR API request failed:{err.response.reason}"
) from err

return response
74 changes: 74 additions & 0 deletions gateway-api/src/gateway_api/test_provider_request.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,74 @@
"""
Unit tests for :mod:`gateway_api.provider_request`.
"""

# imports
import pytest
import requests
from requests import Response
from stubs.stub_provider import GPProviderStub

from gateway_api.provider_request import GPProviderClient

# definitions


# fixtures
@pytest.fixture
def stub() -> GPProviderStub:
return GPProviderStub()


@pytest.fixture
def mock_request_post(monkeypatch: pytest.MonkeyPatch, stub: GPProviderStub) -> None:
"""
Patch requests.post method so calls are routed here.
"""

def _fake_post(
url: str,
headers: dict[str, str], # TODO: define a class 'GPConnectHeaders' for this
timeout: int,
) -> Response:
"""A fake requests.post implementation."""

stub_response = stub.access_record_structured()

return stub_response

monkeypatch.setattr(requests, "post", _fake_post)


# pseudo-code for tests:
# makes valid requests to stub provider and checks responses using a capture

# returns what is received from stub provider (if valid)

# (throws if not 200 OK)
# ~~throws if invalid response from stub provider~~


# receives 200 OK from example.com for valid request
def test_valid_gpprovider_access_structured_record_post_200(
mock_request_post: Response,
stub: GPProviderStub,
) -> None:
"""
Verify that a valid request to the GPProvider returns a 200 OK response.
"""
# Arrange
provider_asid = "200000001154"
consumer_asid = "200000001152"
provider_endpoint = "http://invalid.com"

client = GPProviderClient(
provider_endpoint=provider_endpoint,
provider_asid=provider_asid,
consumer_asid=consumer_asid,
)

# Act
result = client.access_structured_record()

# Assert
assert result.status_code == 200
109 changes: 109 additions & 0 deletions gateway-api/stubs/stubs/stub_provider.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,109 @@
"""
Minimal in-memory stub for a Provider GP System FHIR API,
implementing only accessRecordStructured to read basic
demographic data for a single patient.

Contract elements for direct provider call inferred from from
GPConnect documentation:
https://developer.nhs.uk/apis/gpconnect/accessrecord_structured_development_retrieve_patient_record.html
- Method: POST
- fhir_base: /FHIR/STU3
- resource: /Patient
- fhir_operation: $gpc.getstructruredrecord

Headers:
Ssp-TraceID: Consumer's Trace ID (a GUID or UUID)
Ssp-From: Consumer's ASID
Ssp-To: Provider's ASID
Ssp-InteractionID:
urn:nhs:names:services:gpconnect:fhir:operation:gpc.getstructuredrecord-1

Request Body JSON (FHIR STU3 Parameters resource with patient NHS number.
Later add optional parameters such as `includeAllergies`):
{
"resourceType": "Parameters",
"parameter": [
{
"name": "patientNHSNumber",
"valueIdentifier": {
"system": "https://fhir.nhs.uk/Id/nhs-number",
"value": "9999999999"
}
}
]
}


return

"""

from requests import Response


class GPProviderStub:
"""
A minimal in-memory stub for a Provider GP System FHIR API,
implementing only accessRecordStructured to read basic
demographic data for a single patient.
"""

def __init__(self) -> None:
"""Create a GPProviderStub instance."""
# Seed an example matching the spec's id example stubResponse
# FHIR/STU3 Patient resource with only administrative data based on Example 2
# https://simplifier.net/guide/gp-connect-access-record-structured/Home/Examples/Allergy-examples?version=1.6.2
self.patient_bundle = {
"resourceType": "Bundle",
"type": "collection",
"meta": {
"profile": [
"https://fhir.nhs.uk/STU3/StructureDefinition/GPConnect-StructuredRecord-Bundle-1"
]
},
"entry": [
{
"resource": {
"resourceType": "Patient",
"id": "04603d77-1a4e-4d63-b246-d7504f8bd833",
"meta": {
"versionId": "1469448000000",
"profile": [
"https://fhir.nhs.uk/STU3/StructureDefinition/CareConnect-GPC-Patient-1"
],
},
"identifier": [
{
"system": "https://fhir.nhs.uk/Id/nhs-number",
"value": "9999999999",
}
],
"active": True,
"name": [
{
"use": "official",
"text": "JACKSON Jane (Miss)",
"family": "Jackson",
"given": ["Jane"],
"prefix": ["Miss"],
}
],
"gender": "female",
"birthDate": "1952-05-31",
}
}
],
}

def access_record_structured(self) -> Response:
"""
Simulate accessRecordStructured operation of GPConnect FHIR API.

returns:
Response: The stub patient bundle wrapped in a Response object.
"""

response = Response()
response.status_code = 200

return response
Loading