mesh-sandbox/src/mesh_sandbox/common/exceptions.py at 75b11e7370d23cb665a7c6bceee9993d0cce923d · NHSDigital/mesh-sandbox · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
from typing import Any, Optional, Union
from uuid import uuid4

from ..models.message import Message, MessageEvent, MessageMetadata, MessageParty, MessageStatus, MessageType
from . import constants


class MessagingException(Exception):
    def __init__(
        self,
        status_code: int,
        detail: Any = None,
        headers: Optional[dict[str, Any]] = None,
        message_id: Optional[str] = None,
    ) -> None:
        super().__init__()
        self.status_code = status_code
        self.detail = detail
        self.headers = headers
        self.message_id = message_id


def parse_error(
    detail: Optional[str] = None, message_id: Optional[str] = None, format_params: Optional[tuple] = None
) -> dict:
    if detail is None:
        raise ValueError("error_code or error_description must be supplied")

    error_event, error_code, error_message = constants.ErrorCodeMap.get(detail)  # type: ignore[misc]
    err_response = {
        "errorEvent": error_event,
        "errorCode": error_code,
        "errorDescription": error_message.format(*format_params) if format_params else error_message,
    }
    if message_id:
        err_response["messageID"] = message_id
    return err_response


def try_parse_error(detail: Union[str, dict, None] = None, message_id: Optional[str] = None) -> dict:
    if isinstance(detail, str):
        if detail in constants.ErrorCodeMap:
            return parse_error(detail=detail, message_id=message_id)
        return {"errorDescription": detail}

    if isinstance(detail, dict):
        return detail
    return {"errorDescription": str(detail)}


def get_ndr_error() -> dict:
    expiry_period = 5
    error_description = parse_error(
        detail=constants.ERROR_UNDELIVERED_MESSAGE,
        format_params=(expiry_period,),
    )

    return error_description


def create_ndr(request, recipient) -> Message:
    error_description = get_ndr_error()
    report = create_error_report(request, error_description, recipient)
    return report


def create_error_report(request, error_description: dict, recipient) -> Message:

    error_code = error_description.get("errorCode")
    error_event = error_description.get("errorEvent")
    error_message = error_description.get("errorDescription")

    subject = "NDR" if not request.subject else f"NDR: {request.subject}"

    metadata = MessageMetadata(
        subject=subject,
        local_id=request.local_id,
    )

    return Message(
        events=[
            MessageEvent(status=MessageStatus.ACCEPTED),
            MessageEvent(
                status=MessageStatus.ERROR,
                code=error_code,
                event=error_event,
                description=error_message,
                linked_message_id=request.linked_message_id,
            ),
        ],
        message_id=uuid4().hex.upper(),
        sender=MessageParty(
            mailbox_id="",
            mailbox_name="Central System Mailbox",
            ods_code="X26",
            org_code="X26",
            org_name="NHS England",
            billing_entity="England",
        ),
        recipient=MessageParty(
            mailbox_id=recipient.mailbox_id,
            mailbox_name=recipient.mailbox_name,
            ods_code=recipient.ods_code,
            org_code=recipient.org_code,
            org_name=recipient.org_name,
            billing_entity=recipient.billing_entity,
        ),
        total_chunks=0,
        message_type=MessageType.REPORT,
        workflow_id=request.workflow_id,
        metadata=metadata,
    )