[GPCAPIM-287] Orange box

Makefile

@@ -54,11 +54,21 @@ publish: # Publish the project artefact @Pipeline
 	# TODO: Implement the artefact publishing step
 
 deploy: clean build # Deploy the project artefact to the target environment @Pipeline
-	@if [[ -n "$${IN_BUILD_CONTAINER}" ]]; then \
+	@PROVIDER_STRING="" ; \
+	if [[ -n "$${STUB_PROVIDER}" ]]; then \
+		PROVIDER_STRING="$${PROVIDER_STRING} -e STUB_PROVIDER=$${STUB_PROVIDER}" ; \
+	fi ; \
+	if [[ -n "$${STUB_PDS}" ]]; then \
+		PROVIDER_STRING="$${PROVIDER_STRING} -e STUB_PDS=$${STUB_PDS}" ; \
+	fi ; \
+	if [[ -n "$${STUB_SDS}" ]]; then \
+		PROVIDER_STRING="$${PROVIDER_STRING} -e STUB_SDS=$${STUB_SDS}" ; \
+	fi ; \
+	if [[ -n "$${IN_BUILD_CONTAINER}" ]]; then \
 		echo "Starting using local docker network ..." ; \
-		$(docker) run --platform linux/amd64 --name gateway-api -p 5000:8080 --network gateway-local -d ${IMAGE_NAME} ; \
+		$(docker) run --platform linux/amd64 --name gateway-api -p 5000:8080 --network gateway-local $${PROVIDER_STRING} -d ${IMAGE_NAME} ; \
 	else \
-		$(docker) run --platform linux/amd64 --name gateway-api -p 5000:8080 -d ${IMAGE_NAME} ; \
+		$(docker) run --platform linux/amd64 --name gateway-api -p 5000:8080 $${PROVIDER_STRING} -d ${IMAGE_NAME} ; \
 	fi
 
 clean:: stop # Clean-up project resources (main) @Operations

gateway-api/pyproject.toml

@@ -6,13 +6,14 @@ authors = [
     {name = "Your Name", email = "you@example.com"}
 ]
 readme = "README.md"
-requires-python = ">3.13,<4.0.0"
+requires-python = ">=3.14,<4.0.0"
 
 [tool.poetry.dependencies]
 clinical-data-common = { git = "https://github.com/NHSDigital/clinical-data-common.git", tag = "v0.1.0" }
 flask = "^3.1.2"
 types-flask = "^1.1.6"
 requests = "^2.32.5"
+pyjwt = "^2.11.0"
 
 [tool.poetry]
 packages = [{include = "gateway_api", from = "src"},

gateway-api/scripts/call_gateway.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+
+import argparse
+import json
+import os
+import sys
+from uuid import uuid4
+
+import requests
+
+
+def main() -> None:
+    # Parse command-line arguments
+    parser = argparse.ArgumentParser(
+        description="POST request to GPC getstructuredrecord endpoint"
+    )
+    parser.add_argument(
+        "nhs_number", help="NHS number to search for (e.g., 9690937278)"
+    )
+    args = parser.parse_args()
+
+    # Check if BASE_URL is set
+    base_url = os.environ.get("BASE_URL")
+    if not base_url:
+        print("Error: BASE_URL environment variable is not set")
+        sys.exit(1)
+
+    # Endpoint URL
+    url = f"{base_url}/patient/$gpc.getstructuredrecord"
+
+    # Request headers
+    headers = {
+        "Content-Type": "application/fhir+json",
+        "Accept": "application/fhir+json",
+        "Ssp-TraceID": str(uuid4()),
+        "Ods-From": "S44444",
+    }
+
+    # Request body
+    payload = {
+        "resourceType": "Parameters",
+        "parameter": [
+            {
+                "name": "patientNHSNumber",
+                "valueIdentifier": {
+                    "system": "https://fhir.nhs.uk/Id/nhs-number",
+                    "value": args.nhs_number,
+                },
+            }
+        ],
+    }
+
+    # Make the POST request
+    try:
+        response = requests.post(url, headers=headers, json=payload, timeout=10)
+        response.raise_for_status()
+
+        print(f"Status Code: {response.status_code}")
+        print(f"Response:\n{json.dumps(response.json(), indent=2)}")
+
+    except requests.exceptions.RequestException as e:
+        errtext = f"Error: {e}\n"
+        if e.response is not None:
+            errtext += f"Status Code: {e.response.status_code}\n"
+            errtext += f"Response Body: {e.response.text}\n"
+        print(errtext)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

gateway-api/src/gateway_api/clinical_jwt/__init__.py

@@ -0,0 +1,6 @@
+from .device import Device
+from .jwt import JWT
+from .organization import Organization
+from .practitioner import Practitioner
+
+__all__ = ["JWT", "Device", "Organization", "Practitioner"]

gateway-api/src/gateway_api/clinical_jwt/device.py

@@ -0,0 +1,34 @@
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass(frozen=True, kw_only=True)
+class Device:
+    system: str
+    value: str
+    model: str
+    version: str
+
+    def to_dict(self) -> dict[str, Any]:
+        """
+        Return the Device as a dictionary suitable for JWT payload.
+        """
+        return {
+            "resourceType": "Device",
+            "identifier": [{"system": self.system, "value": self.value}],
+            "model": self.model,
+            "version": self.version,
+        }
+
+    @property
+    def json(self) -> dict[str, Any]:
+        """
+        Return the Device as a dictionary suitable for JWT payload.
+        Provided for backwards compatibility.
+        """
+        return self.to_dict()
+
+    def __str__(self) -> str:
+        import json
+
+        return json.dumps(self.to_dict(), indent=2)

gateway-api/src/gateway_api/clinical_jwt/jwt.py

@@ -0,0 +1,77 @@
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from time import time
+from typing import Any
+
+import jwt as pyjwt
+
+
+@dataclass(frozen=True, kw_only=True)
+class JWT:
+    issuer: str
+    subject: str
+    audience: str
+    requesting_device: dict[str, Any]
+    requesting_organization: dict[str, Any]
+    requesting_practitioner: dict[str, Any]
+
+    # Time fields
+    issued_at: int = field(default_factory=lambda: int(time()))
+    expiration: int = field(default_factory=lambda: int(time()) + 300)
+
+    # These are here for future proofing but are not expected ever to be changed
+    algorithm: str | None = None
+    type: str = "JWT"
+    reason_for_request: str = "directcare"
+    requested_scope: str = "patient/*.read"
+
+    @property
+    def issue_time(self) -> str:
+        return datetime.fromtimestamp(self.issued_at, tz=UTC).isoformat()
+
+    @property
+    def exp_time(self) -> str:
+        return datetime.fromtimestamp(self.expiration, tz=UTC).isoformat()
+
+    def encode(self) -> str:
+        return pyjwt.encode(
+            self.payload(),
+            key=None,
+            algorithm=self.algorithm,
+            headers={"typ": self.type},
+        )
+
+    @staticmethod
+    def decode(token: str) -> "JWT":
+        token_dict = pyjwt.decode(
+            token,
+            options={"verify_signature": False},  # NOSONAR S5659 (not signed)
+        )
+
+        return JWT(
+            issuer=token_dict["iss"],
+            subject=token_dict["sub"],
+            audience=token_dict["aud"],
+            expiration=token_dict["exp"],
+            issued_at=token_dict["iat"],
+            requesting_device=token_dict["requesting_device"],
+            requesting_organization=token_dict["requesting_organization"],
+            requesting_practitioner=token_dict["requesting_practitioner"],
+        )
+
+    def payload(self) -> dict[str, Any]:
+        return {
+            "iss": self.issuer,
+            "sub": self.subject,
+            "aud": self.audience,
+            "exp": self.expiration,
+            "iat": self.issued_at,
+            "requesting_device": self.requesting_device,
+            "requesting_organization": self.requesting_organization,
+            "requesting_practitioner": self.requesting_practitioner,
+            "reason_for_request": self.reason_for_request,
+            "requested_scope": self.requested_scope,
+        }
+
+    def __str__(self) -> str:
+        return self.encode()

gateway-api/src/gateway_api/clinical_jwt/organization.py

@@ -0,0 +1,36 @@
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass(frozen=True, kw_only=True)
+class Organization:
+    ods_code: str
+    name: str
+
+    def to_dict(self) -> dict[str, Any]:
+        """
+        Return the Organization as a dictionary suitable for JWT payload.
+        """
+        return {
+            "resourceType": "Organization",
+            "identifier": [
+                {
+                    "system": "https://fhir.nhs.uk/Id/ods-organization-code",
+                    "value": self.ods_code,
+                }
+            ],
+            "name": self.name,
+        }
+
+    @property
+    def json(self) -> dict[str, Any]:
+        """
+        Return the Organization as a dictionary suitable for JWT payload.
+        Provided for backwards compatibility.
+        """
+        return self.to_dict()
+
+    def __str__(self) -> str:
+        import json
+
+        return json.dumps(self.to_dict(), indent=2)

gateway-api/src/gateway_api/clinical_jwt/practitioner.py

@@ -0,0 +1,51 @@
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass(kw_only=True)
+class Practitioner:
+    id: str
+    sds_userid: str
+    role_profile_id: str
+    userid_url: str
+    userid_value: str
+    family_name: str
+    given_name: str | None = None
+    prefix: str | None = None
+
+    def to_dict(self) -> dict[str, Any]:
+        """
+        Return the Practitioner as a dictionary suitable for JWT payload.
+        """
+        user_id_system = "https://fhir.nhs.uk/Id/sds-user-id"
+        role_id_system = "https://fhir.nhs.uk/Id/sds-role-profile-id"
+
+        name_dict: dict[str, Any] = {"family": self.family_name}
+        if self.given_name is not None:
+            name_dict["given"] = [self.given_name]
+        if self.prefix is not None:
+            name_dict["prefix"] = [self.prefix]
+
+        return {
+            "resourceType": "Practitioner",
+            "id": self.id,
+            "identifier": [
+                {"system": user_id_system, "value": self.sds_userid},
+                {"system": role_id_system, "value": self.role_profile_id},
+                {"system": self.userid_url, "value": self.userid_value},
+            ],
+            "name": [name_dict],
+        }
+
+    @property
+    def json(self) -> dict[str, Any]:
+        """
+        Return the Practitioner as a dictionary suitable for JWT payload.
+        Provided for backwards compatibility.
+        """
+        return self.to_dict()
+
+    def __str__(self) -> str:
+        import json
+
+        return json.dumps(self.to_dict(), indent=2)