ELI-138 Audit record within lambda

src/eligibility_signposting_api/audit_context.py

@@ -0,0 +1,142 @@
+import logging
+from dataclasses import asdict
+from datetime import UTC, datetime
+from operator import attrgetter
+
+from flask import Request, g
+
+from eligibility_signposting_api.audit_models import (
+    AuditAction,
+    AuditCondition,
+    AuditEligibilityCohortGroups,
+    AuditEligibilityCohorts,
+    AuditEvent,
+    AuditFilterRule,
+    AuditRedirectRule,
+    AuditSuitabilityRule,
+    RequestAuditData,
+    RequestAuditHeader,
+    RequestAuditQueryParams,
+)
+from eligibility_signposting_api.model.eligibility import (
+    CohortGroupResult,
+    ConditionName,
+    IterationResult,
+    Status,
+    SuggestedActions,
+)
+from eligibility_signposting_api.model.rules import CampaignID, CampaignVersion, Iteration, RuleName, RulePriority
+from eligibility_signposting_api.services.audit_service import AuditService
+
+logger = logging.getLogger(__name__)
+
+
+class AuditContext:
+    @staticmethod
+    def add_request_details(request: Request) -> None:
+        g.audit_log = AuditEvent()
+        resource_id = None
+        if "nhs_number" in request.view_args:
+            resource_id = request.view_args["nhs_number"]
+        g.audit_log.request = RequestAuditData(
+            nhs_number=resource_id,
+            request_timestamp=datetime.now(tz=UTC),
+            headers=(
+                RequestAuditHeader(
+                    x_request_id=request.headers.get("X-Request-ID"),
+                    x_correlation_id=request.headers.get("X-Correlation-ID"),
+                    nhsd_end_user_organisation_ods=request.headers.get("NHSD-End-User-Organisation-ODS"),
+                    nhsd_application_id=request.headers.get("nhsd-application-id"),
+                )
+            ),
+            query_params=(
+                RequestAuditQueryParams(
+                    category=request.args.get("category"),
+                    conditions=request.args.get("conditions"),
+                    include_actions=request.args.get("includeActions"),
+                )
+            ),
+        )
+
+    @staticmethod
+    def append_audit_condition(
+        suggested_actions: SuggestedActions | None,
+        condition_name: ConditionName,
+        best_results: tuple[Iteration, IterationResult, dict[str, CohortGroupResult]],
+        campaign_details: tuple[CampaignID | None, CampaignVersion | None],
+        redirect_rule_details: tuple[RulePriority | None, RuleName | None],
+    ) -> None:
+        audit_eligibility_cohorts, audit_eligibility_cohort_groups, audit_actions = [], [], []
+        audit_filter_rule, audit_suitability_rule, audit_redirect_rule = None, None, None
+        best_active_iteration = best_results[0]
+        best_candidate = best_results[1]
+        best_cohort_results = best_results[2]
+
+        for value in sorted(best_cohort_results.values(), key=attrgetter("cohort_code")):
+            audit_eligibility_cohorts.append(
+                AuditEligibilityCohorts(cohort_code=value.cohort_code, cohort_status=value.status.name)
+            )
+
+            audit_eligibility_cohort_groups.append(
+                AuditEligibilityCohortGroups(
+                    cohort_code=value.cohort_code, cohort_status=value.status.name, cohort_text=value.description
+                )
+            )
+
+            if value.audit_rules:
+                if best_candidate.status.name == Status.not_eligible.name:
+                    audit_filter_rule = AuditFilterRule(
+                        rule_priority=int(value.audit_rules[0].rule_priority), rule_name=value.audit_rules[0].rule_name
+                    )
+                if best_candidate.status.name == Status.not_actionable.name:
+                    audit_suitability_rule = AuditSuitabilityRule(
+                        rule_priority=int(value.audit_rules[0].rule_priority),
+                        rule_name=value.audit_rules[0].rule_name,
+                        rule_message=value.audit_rules[0].rule_description,
+                    )
+
+        if best_candidate.status.name == Status.actionable.name:
+            audit_redirect_rule = AuditRedirectRule(redirect_rule_details[0], redirect_rule_details[1])
+
+        if suggested_actions is None or suggested_actions == []:
+            audit_actions = suggested_actions
+
+        elif len(suggested_actions.actions) > 0:
+            for action in suggested_actions.actions:
+                audit_actions.append(
+                    AuditAction(
+                        action_code=action.action_code,
+                        action_type=action.action_type,
+                        action_description=action.action_description,
+                        action_url=action.url_link,
+                        action_url_label=action.url_label,
+                    )
+                )
+
+        audit_condition = AuditCondition(
+            campaign_id=campaign_details[0],
+            campaign_version=campaign_details[1],
+            iteration_id=best_active_iteration.id,
+            iteration_version=best_active_iteration.version,
+            condition_name=condition_name,
+            status=best_candidate.status.name,
+            status_text=best_candidate.status.name,
+            eligibility_cohorts=audit_eligibility_cohorts,
+            eligibility_cohort_groups=audit_eligibility_cohort_groups,
+            filter_rules=audit_filter_rule,
+            suitability_rules=audit_suitability_rule,
+            action_rule=audit_redirect_rule,
+            actions=audit_actions,
+        )
+
+        g.audit_log.response.condition.append(audit_condition)
+
+    @staticmethod
+    def add_response_details(response) -> None:
+        g.audit_log.response.response_id = response.response_id
+        g.audit_log.response.last_updated = response.meta.last_updated
+
+    @staticmethod
+    def write_to_firehose(service: AuditService, response) -> None:
+        AuditContext.add_response_details(response)
+        service.audit(asdict(g.audit_log))

src/eligibility_signposting_api/audit_models.py

@@ -0,0 +1,98 @@
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from uuid import UUID
+
+
+@dataclass
+class RequestAuditHeader:
+    x_request_id: str = None
+    x_correlation_id: str = None
+    nhsd_end_user_organisation_ods: str = None
+    nhsd_application_id: str = None
+
+
+@dataclass
+class RequestAuditQueryParams:
+    category: str | None = None
+    conditions: str | None = None
+    include_actions: str | None = None
+
+
+@dataclass
+class RequestAuditData:
+    request_timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
+    headers: RequestAuditHeader = field(default_factory=RequestAuditHeader)
+    query_params: RequestAuditQueryParams = field(default_factory=RequestAuditQueryParams)
+    nhs_number: str | None = None
+
+
+@dataclass
+class AuditEligibilityCohorts:
+    cohort_code: str | None = None
+    cohort_status: str | None = None
+
+
+@dataclass
+class AuditEligibilityCohortGroups:
+    cohort_code: str | None = None
+    cohort_text: str | None = None
+    cohort_status: str | None = None
+
+
+@dataclass
+class AuditFilterRule:
+    rule_priority: int | None = None
+    rule_name: str | None = None
+
+
+@dataclass
+class AuditSuitabilityRule:
+    rule_priority: int | None = None
+    rule_name: str | None = None
+    rule_message: str | None = None
+
+
+@dataclass
+class AuditRedirectRule:
+    rule_priority: int | None = None
+    rule_name: str | None = None
+
+
+@dataclass
+class AuditAction:
+    internal_name: str | None = None
+    action_type: str | None = None
+    action_code: str | None = None
+    action_description: str | None = None
+    action_url: str | None = None
+    action_url_label: str | None = None
+
+
+@dataclass
+class AuditCondition:
+    campaign_id: str = None
+    campaign_version: str = None
+    iteration_id: str = None
+    iteration_version: str = None
+    condition_name: str = None
+    status: str = None
+    status_text: str = None
+    eligibility_cohorts: list[AuditEligibilityCohorts] | None = None
+    eligibility_cohort_groups: list[AuditEligibilityCohortGroups] | None = None
+    filter_rules: AuditFilterRule | None = None
+    suitability_rules: AuditSuitabilityRule | None = None
+    action_rule: AuditRedirectRule | None = None
+    actions: list[AuditAction] = field(default_factory=list)
+
+
+@dataclass
+class ResponseAuditData:
+    response_id: UUID | None = None
+    last_updated: str | None = None
+    condition: list[AuditCondition] = field(default_factory=list)
+
+
+@dataclass
+class AuditEvent:
+    request: RequestAuditData = field(default_factory=RequestAuditData)
+    response: ResponseAuditData = field(default_factory=ResponseAuditData)

src/eligibility_signposting_api/model/eligibility.py

@@ -13,6 +13,7 @@
 
 RuleName = NewType("RuleName", str)
 RuleDescription = NewType("RuleDescription", str)
+RulePriority = NewType("RulePriority", str)
 
 ActionType = NewType("ActionType", str)
 ActionCode = NewType("ActionCode", str)
@@ -66,6 +67,7 @@ def best(*statuses: Status) -> Status:
 class Reason:
     rule_type: RuleType
     rule_name: RuleName
+    rule_priority: RulePriority
     rule_description: RuleDescription | None
     matcher_matched: bool
 
@@ -98,6 +100,7 @@ class CohortGroupResult:
     status: Status
     reasons: list[Reason]
     description: str | None
+    audit_rules: list[Reason]
 
 
 @dataclass

src/eligibility_signposting_api/services/audit_service.py

@@ -31,8 +31,9 @@ def audit(self, audit_record: dict) -> None:
         Returns:
             str: The Firehose record ID.
         """
+        data = json.dumps(audit_record, default=str)
         response = self.firehose.put_record(
             DeliveryStreamName=self.audit_delivery_stream,
-            Record={"Data": (json.dumps(audit_record) + "\n").encode("utf-8")},
+            Record={"Data": (data + "\n").encode("utf-8")},
         )
         logger.info("Successfully sent to the Firehose", extra={"firehose_record_id": response["RecordId"]})