airflow_scheduler/.venv/Lib/site-packages/neo4j/_work/summary.py

# Copyright (c) "Neo4j"
# Neo4j Sweden AB [https://neo4j.com]
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


from __future__ import annotations

import itertools
import typing as t
from copy import deepcopy
from dataclasses import dataclass

from .._api import (
    NotificationCategory,
    NotificationClassification,
    NotificationSeverity,
)
from .._exceptions import BoltProtocolError
from .._meta import preview


if t.TYPE_CHECKING:
    import typing_extensions as te

    from ..addressing import Address
    from ..api import ServerInfo

    _T = te.TypeVar("_T")


class ResultSummary:
    """A summary of execution returned with a :class:`.Result` object."""

    #: A :class:`neo4j.ServerInfo` instance. Provides some basic information of
    #: the server where the result is obtained from.
    server: ServerInfo

    #: The database name where this summary is obtained from.
    database: str | None

    #: The query that was executed to produce this result.
    query: str | None

    #: Dictionary of parameters passed with the statement.
    parameters: dict[str, t.Any] | None

    #: A string that describes the type of query
    # ``'r'`` = read-only, ``'rw'`` = read/write, ``'w'`` = write-only,
    # ``'s'`` = schema.
    query_type: te.Literal["r", "rw", "w", "s"] | None

    #: A :class:`neo4j.SummaryCounters` instance. Counters for operations the
    #: query triggered.
    counters: SummaryCounters

    #: Dictionary that describes how the database will execute the query.
    plan: dict | None

    #: Dictionary that describes how the database executed the query.
    profile: dict | None

    #: The time it took for the server to have the result available.
    #: (milliseconds)
    result_available_after: int | None

    #: The time it took for the server to consume the result. (milliseconds)
    result_consumed_after: int | None

    #: A list of Dictionaries containing notification information.
    #: Notifications provide extra information for a user executing a
    #: statement.
    #: They can be warnings about problematic queries or other valuable
    #: information that can be
    #: presented in a client.
    #: Unlike failures or errors, notifications do not affect the execution of
    #: a statement.
    #:
    #: .. seealso:: :attr:`.summary_notifications`
    notifications: list[dict] | None

    # cache for notifications
    _notifications_set: bool = False

    # cache for property `summary_notifications`
    _summary_notifications: list[SummaryNotification]

    # cache for property `summary_notifications`
    _gql_status_objects: tuple[GqlStatusObject, ...]

    _had_key: bool
    _had_record: bool

    def __init__(
        self,
        address: Address,
        had_key: bool,
        had_record: bool,
        metadata: dict[str, t.Any],
    ) -> None:
        self._had_key = had_key
        self._had_record = had_record
        self.metadata = metadata
        self.server = metadata["server"]
        self.database = metadata.get("db")
        self.query = metadata.get("query")
        self.parameters = metadata.get("parameters")
        if "type" in metadata:
            self.query_type = metadata["type"]
            if self.query_type not in {"r", "w", "rw", "s"}:
                raise BoltProtocolError(
                    f"Unexpected query type {self.query_type!r} received from "
                    "server. Consider updating the driver.",
                    address,
                )
        self.query_type = metadata.get("type")
        self.plan = metadata.get("plan")
        self.profile = metadata.get("profile")
        self.counters = SummaryCounters(metadata.get("stats", {}))
        if self.server.protocol_version[0] < 3:
            self.result_available_after = metadata.get(
                "result_available_after"
            )
            self.result_consumed_after = metadata.get("result_consumed_after")
        else:
            self.result_available_after = metadata.get("t_first")
            self.result_consumed_after = metadata.get("t_last")

    def __dir__(self):
        return {*super().__dir__(), "notifications"}

    def __getattr__(self, key):
        if key == "notifications":
            self._set_notifications()
            return self.notifications
        raise AttributeError(
            f"'{self.__class__.__name__}' object has no attribute '{key}'"
        )

    @staticmethod
    def _notification_from_status(status: dict) -> dict:
        notification = {}
        for notification_key, status_key in (
            ("title", "title"),
            ("code", "neo4j_code"),
            ("description", "description"),
        ):
            if status_key in status:
                notification[notification_key] = status[status_key]

        if "diagnostic_record" in status:
            diagnostic_record = status["diagnostic_record"]
            if not isinstance(diagnostic_record, dict):
                diagnostic_record = {}

            for notification_key, diag_record_key in (
                ("severity", "_severity"),
                ("category", "_classification"),
                ("position", "_position"),
            ):
                if diag_record_key in diagnostic_record:
                    notification[notification_key] = diagnostic_record[
                        diag_record_key
                    ]

        return notification

    def _set_notifications(self) -> None:
        if "notifications" in self.metadata:
            notifications = self.metadata["notifications"]
            if not isinstance(notifications, list):
                self.notifications = None
                return
            self.notifications = notifications
            return

        # polyfill notifications from GqlStatusObjects
        if "statuses" in self.metadata:
            statuses = self.metadata["statuses"]
            if not isinstance(statuses, list):
                self.notifications = None
                return
            notifications = []
            for status in statuses:
                if not (isinstance(status, dict) and "neo4j_code" in status):
                    # not a notification status
                    continue
                notification = self._notification_from_status(status)
                notifications.append(notification)
            self.notifications = notifications or None
            return

        self.notifications = None

    # TODO: 6.0 - return a tuple for immutability (annotate with Sequence)
    @property
    def summary_notifications(self) -> list[SummaryNotification]:
        """
        The same as ``notifications`` but in a parsed, structured form.

        Further, if connected to a gql-aware server, this property will be
        polyfilled from :attr:`gql_status_objects`.

        .. seealso:: :attr:`.notifications`, :class:`.SummaryNotification`

        .. versionadded:: 5.7
        """
        if getattr(self, "_summary_notifications", None) is not None:
            return self._summary_notifications

        raw_notifications = self.notifications
        if not isinstance(raw_notifications, list):
            self._summary_notifications = []
            return self._summary_notifications
        self._summary_notifications = [
            SummaryNotification._from_metadata(n) for n in raw_notifications
        ]
        return self._summary_notifications

    @property
    @preview("GQLSTATUS support is a preview feature.")
    def gql_status_objects(self) -> t.Sequence[GqlStatusObject]:
        """
        Get GqlStatusObjects that arose when executing the query.

        The sequence always contains at least 1 status representing the
        Success, No Data or Omitted Result.
        All other status are notifications like warnings about problematic
        queries or other valuable information that can be presented in a
        client.

        The GqlStatusObjects will be presented in the following order:

        * A "no data" (``02xxx``) has precedence over a warning.
        * A "warning" (``01xxx``) has precedence over a success.
        * A "success" (``00xxx``) has precedence over anything informational
          (``03xxx``).

        **This is a preview** (see :ref:`filter-warnings-ref`).
        It might be changed without following the deprecation policy.

        See also
        https://github.com/neo4j/neo4j-python-driver/wiki/preview-features

        .. versionadded:: 5.22
        """
        raw_status_objects = self.metadata.get("statuses")
        if isinstance(raw_status_objects, list):
            self._gql_status_objects = tuple(
                GqlStatusObject._from_status_metadata(s)
                for s in raw_status_objects
            )
            return self._gql_status_objects

        raw_notifications = self.notifications
        notification_status_objects: t.Iterable[GqlStatusObject]
        if isinstance(raw_notifications, list):
            notification_status_objects = [
                GqlStatusObject._from_notification_metadata(n)
                for n in raw_notifications
            ]
        else:
            notification_status_objects = ()

        if self._had_record:
            # polyfill with a Success status
            result_status = GqlStatusObject._success()
        elif self._had_key:
            # polyfill with an Omitted Result status
            result_status = GqlStatusObject._no_data()
        else:
            # polyfill with a No Data status
            result_status = GqlStatusObject._omitted_result()

        notification_status_objects = itertools.chain(
            notification_status_objects, (result_status,)
        )

        def status_precedence(status: GqlStatusObject) -> int:
            if status.gql_status.startswith("02"):
                # no data
                return 3
            if status.gql_status.startswith("01"):
                # warning
                return 2
            if status.gql_status.startswith("00"):
                # success
                return 1
            if status.gql_status.startswith("03"):
                # informational
                return 0
            return -1

        notification_status_objects = sorted(
            notification_status_objects,
            key=status_precedence,
            reverse=True,
        )
        self._gql_status_objects = tuple(notification_status_objects)

        return self._gql_status_objects


class SummaryCounters:
    """Contains counters for various operations that a query triggered."""

    #:
    nodes_created: int = 0

    #:
    nodes_deleted: int = 0

    #:
    relationships_created: int = 0

    #:
    relationships_deleted: int = 0

    #:
    properties_set: int = 0

    #:
    labels_added: int = 0

    #:
    labels_removed: int = 0

    #:
    indexes_added: int = 0

    #:
    indexes_removed: int = 0

    #:
    constraints_added: int = 0

    #:
    constraints_removed: int = 0

    #:
    system_updates: int = 0

    _contains_updates = None
    _contains_system_updates = None

    def __init__(self, statistics) -> None:
        key_to_attr_name = {
            "nodes-created": "nodes_created",
            "nodes-deleted": "nodes_deleted",
            "relationships-created": "relationships_created",
            "relationships-deleted": "relationships_deleted",
            "properties-set": "properties_set",
            "labels-added": "labels_added",
            "labels-removed": "labels_removed",
            "indexes-added": "indexes_added",
            "indexes-removed": "indexes_removed",
            "constraints-added": "constraints_added",
            "constraints-removed": "constraints_removed",
            "system-updates": "system_updates",
            "contains-updates": "_contains_updates",
            "contains-system-updates": "_contains_system_updates",
        }
        for key, value in dict(statistics).items():
            attr_name = key_to_attr_name.get(key)
            if attr_name:
                setattr(self, attr_name, value)

    def __repr__(self) -> str:
        return repr(vars(self))

    @property
    def contains_updates(self) -> bool:
        """
        Check if any counters tracking graph updates are greater than 0.

        True if any of the counters except for system_updates, are greater
        than 0. Otherwise, False.
        """
        if self._contains_updates is not None:
            return self._contains_updates
        return bool(
            self.nodes_created
            or self.nodes_deleted
            or self.relationships_created
            or self.relationships_deleted
            or self.properties_set
            or self.labels_added
            or self.labels_removed
            or self.indexes_added
            or self.indexes_removed
            or self.constraints_added
            or self.constraints_removed
        )

    @property
    def contains_system_updates(self) -> bool:
        """True if the system database was updated, otherwise False."""
        if self._contains_system_updates is not None:
            return self._contains_system_updates
        return self.system_updates > 0


@dataclass
class SummaryInputPosition:
    """
    Structured form of a gql status/notification position.

    .. seealso::
        :attr:`.GqlStatusObject.position`,
        :attr:`.SummaryNotification.position`,
        :data:`.SummaryNotificationPosition`

    .. versionadded:: 5.22
    """

    #: The line number of the notification. Line numbers start at 1.
    line: int
    #: The column number of the notification. Column numbers start at 1.
    column: int
    #: The character offset of the notification. Offsets start at 0.
    offset: int

    @classmethod
    def _from_metadata(cls, metadata: object) -> te.Self | None:
        if not isinstance(metadata, dict):
            return None
        line = metadata.get("line")
        if not isinstance(line, int) or isinstance(line, bool):
            return None
        column = metadata.get("column")
        if not isinstance(column, int) or isinstance(column, bool):
            return None
        offset = metadata.get("offset")
        if not isinstance(offset, int) or isinstance(offset, bool):
            return None
        return cls(line=line, column=column, offset=offset)

    def __str__(self) -> str:
        return (
            f"line: {self.line}, column: {self.column}, offset: {self.offset}"
        )


# Deprecated alias for :class:`.SummaryInputPosition`.
#
# .. versionadded:: 5.7
#
# .. versionchanged:: 5.22
#     Deprecated in favor of :class:`.SummaryInputPosition`.
SummaryNotificationPosition: te.TypeAlias = SummaryInputPosition


_SEVERITY_LOOKUP: dict[t.Any, NotificationSeverity] = {
    "WARNING": NotificationSeverity.WARNING,
    "INFORMATION": NotificationSeverity.INFORMATION,
}

_CATEGORY_LOOKUP: dict[t.Any, NotificationCategory] = {
    "HINT": NotificationCategory.HINT,
    "UNRECOGNIZED": NotificationCategory.UNRECOGNIZED,
    "UNSUPPORTED": NotificationCategory.UNSUPPORTED,
    "PERFORMANCE": NotificationCategory.PERFORMANCE,
    "DEPRECATION": NotificationCategory.DEPRECATION,
    "GENERIC": NotificationCategory.GENERIC,
    "SECURITY": NotificationCategory.SECURITY,
    "TOPOLOGY": NotificationCategory.TOPOLOGY,
    "SCHEMA": NotificationCategory.SCHEMA,
}

_CLASSIFICATION_LOOKUP: dict[t.Any, NotificationClassification] = {
    k: NotificationClassification(v) for k, v in _CATEGORY_LOOKUP.items()
}


if t.TYPE_CHECKING:

    class _SummaryNotificationKwargs(te.TypedDict, total=False):
        title: str
        code: str
        description: str
        severity_level: NotificationSeverity
        category: NotificationCategory
        raw_severity_level: str
        raw_category: str
        position: SummaryInputPosition | None


@dataclass
class SummaryNotification:
    """
    Structured form of a notification received from the server.

    .. seealso:: :attr:`.ResultSummary.summary_notifications`

    .. versionadded:: 5.7
    """

    title: str = ""
    code: str = ""
    description: str = ""
    severity_level: NotificationSeverity = NotificationSeverity.UNKNOWN
    category: NotificationCategory = NotificationCategory.UNKNOWN
    raw_severity_level: str = ""
    raw_category: str = ""
    position: SummaryNotificationPosition | None = None

    @classmethod
    def _from_metadata(cls, metadata: object) -> te.Self:
        if not isinstance(metadata, dict):
            return cls()
        kwargs: _SummaryNotificationKwargs = {
            "position": SummaryInputPosition._from_metadata(
                metadata.get("position")
            ),
        }
        str_keys: tuple[te.Literal["title", "code", "description"], ...] = (
            "title",
            "code",
            "description",
        )
        for key in str_keys:
            value = metadata.get(key)
            if isinstance(value, str):
                kwargs[key] = value
        severity = metadata.get("severity")
        if isinstance(severity, str):
            kwargs["raw_severity_level"] = severity
            kwargs["severity_level"] = _SEVERITY_LOOKUP.get(
                severity, NotificationSeverity.UNKNOWN
            )
        category = metadata.get("category")
        if isinstance(category, str):
            kwargs["raw_category"] = category
            kwargs["category"] = _CATEGORY_LOOKUP.get(
                category, NotificationCategory.UNKNOWN
            )
        return cls(**kwargs)

    def __str__(self) -> str:
        return (
            f"{{severity: {self.raw_severity_level}}} {{code: {self.code}}} "
            f"{{category: {self.raw_category}}} {{title: {self.title}}} "
            f"{{description: {self.description}}} "
            f"{{position: {self.position}}}"
        )


POLYFILL_DIAGNOSTIC_RECORD = (
    ("OPERATION", ""),
    ("OPERATION_CODE", "0"),
    ("CURRENT_SCHEMA", "/"),
)


_SUCCESS_STATUS_METADATA = {
    "gql_status": "00000",
    "status_description": "note: successful completion",
    "diagnostic_record": dict(POLYFILL_DIAGNOSTIC_RECORD),
}
_OMITTED_RESULT_STATUS_METADATA = {
    "gql_status": "00001",
    "status_description": "note: successful completion - omitted result",
    "diagnostic_record": dict(POLYFILL_DIAGNOSTIC_RECORD),
}
_NO_DATA_STATUS_METADATA = {
    "gql_status": "02000",
    "status_description": "note: no data",
    "diagnostic_record": dict(POLYFILL_DIAGNOSTIC_RECORD),
}


class GqlStatusObject:
    """
    Representation for GqlStatusObject found when executing a query.

    GqlStatusObjects are a superset of notifications, i.e., some but not all
    GqlStatusObjects are notifications.
    Notifications can be filtered server-side with
    driver config
    :ref:`driver-notifications-disabled-classifications-ref` and
    :ref:`driver-notifications-min-severity-ref` as well as
    session config
    :ref:`session-notifications-disabled-classifications-ref` and
    :ref:`session-notifications-min-severity-ref`.

    .. seealso:: :attr:`.ResultSummary.gql_status_objects`

    .. versionadded:: 5.22
    """

    # internal dictionaries, never handed to assure immutability
    _status_metadata: dict[str, t.Any]
    _status_diagnostic_record: dict[str, t.Any] | None = None

    _is_notification: bool
    _gql_status: str
    _status_description: str
    _position: SummaryInputPosition | None
    _raw_classification: str | None
    _classification: NotificationClassification
    _raw_severity: str | None
    _severity: NotificationSeverity
    _diagnostic_record: dict[str, t.Any]

    @classmethod
    def _success(cls) -> te.Self:
        obj = cls()
        obj._status_metadata = _SUCCESS_STATUS_METADATA
        return obj

    @classmethod
    def _omitted_result(cls) -> te.Self:
        obj = cls()
        obj._status_metadata = _OMITTED_RESULT_STATUS_METADATA
        return obj

    @classmethod
    def _no_data(cls) -> te.Self:
        obj = cls()
        obj._status_metadata = _NO_DATA_STATUS_METADATA
        return obj

    @classmethod
    def _from_status_metadata(cls, metadata: object) -> te.Self:
        obj = cls()
        if isinstance(metadata, dict):
            obj._status_metadata = metadata
        else:
            obj._status_metadata = {}
        return obj

    @classmethod
    def _from_notification_metadata(cls, metadata: object) -> te.Self:
        obj = cls()
        if not isinstance(metadata, dict):
            metadata = {}
        description = metadata.get("description")
        neo4j_code = metadata.get("neo4j_code")
        if not isinstance(neo4j_code, str):
            neo4j_code = ""
        title = metadata.get("title")
        if not isinstance(title, str):
            title = ""
        position = SummaryInputPosition._from_metadata(
            metadata.get("position")
        )
        classification = metadata.get("category")
        if not isinstance(classification, str):
            classification = None
        severity = metadata.get("severity")
        if not isinstance(severity, str):
            severity = None

        if severity == "WARNING":
            gql_status = "01N42"
            if not isinstance(description, str) or not description:
                description = "warn: unknown warning"
        else:
            # for "INFORMATION" or if severity is missing
            gql_status = "03N42"
            if not isinstance(description, str) or not description:
                description = "info: unknown notification"

        diagnostic_record = dict(POLYFILL_DIAGNOSTIC_RECORD)
        if "category" in metadata:
            diagnostic_record["_classification"] = metadata["category"]
        if "severity" in metadata:
            diagnostic_record["_severity"] = metadata["severity"]
        if "position" in metadata:
            diagnostic_record["_position"] = metadata["position"]

        obj._status_metadata = {
            "gql_status": gql_status,
            "status_description": description,
            "neo4j_code": neo4j_code,
            "title": title,
            "diagnostic_record": diagnostic_record,
        }
        obj._gql_status = gql_status
        obj._status_description = description
        obj._position = position
        obj._raw_classification = classification
        obj._raw_severity = severity
        obj._is_notification = True
        return obj

    def __str__(self) -> str:
        return self.status_description

    def __repr__(self) -> str:
        return (
            "GqlStatusObject("
            f"gql_status={self.gql_status!r}, "
            f"status_description={self.status_description!r}, "
            f"position={self.position!r}, "
            f"raw_classification={self.raw_classification!r}, "
            f"classification={self.classification!r}, "
            f"raw_severity={self.raw_severity!r}, "
            f"severity={self.severity!r}, "
            f"diagnostic_record={self.diagnostic_record!r}"
            ")"
        )

    @property
    def is_notification(self) -> bool:
        """
        Whether this GqlStatusObject is a notification.

        Only some GqlStatusObjects are notifications.
        The definition of notification is vendor-specific.
        Notifications are those GqlStatusObjects that provide additional
        information and can be filtered out via
        :ref:`driver-notifications-disabled-classifications-ref` and
        :ref:`driver-notifications-min-severity-ref` as well as.

        The fields :attr:`.position`,
        :attr:`.raw_classification`, :attr:`.classification`,
        :attr:`.raw_severity`, and :attr:`.severity` are only meaningful
        for notifications.
        """
        if hasattr(self, "_is_notification"):
            return self._is_notification

        neo4j_code = self._status_metadata.get("neo4j_code")
        self._is_notification = bool(
            isinstance(neo4j_code, str) and neo4j_code
        )
        return self._is_notification

    @classmethod
    def _extract_str_field(
        cls,
        data: dict[str, t.Any],
        key: str,
        default: _T = "",  # type: ignore[assignment]
    ) -> str | _T:
        value = data.get(key)
        if isinstance(value, str):
            return value
        else:
            return default

    @property
    def gql_status(self) -> str:
        """
        The GQLSTATUS.

        The following GQLSTATUS codes denote codes that the driver will use
        for polyfilling (when connected to an old, non-GQL-aware server).
        Further, they may be used by servers during the transition-phase to
        GQLSTATUS-awareness.

         * ``01N42`` (warning - unknown warning)
         * ``02N42`` (no data - unknown subcondition)
         * ``03N42`` (informational - unknown notification)
         * ``05N42`` (general processing exception - unknown error)

        .. note::
            This means these codes are not guaranteed to be stable and may
            change in future versions of the driver or the server.
        """
        if hasattr(self, "_gql_status"):
            return self._gql_status

        self._gql_status = self._extract_str_field(
            self._status_metadata, "gql_status"
        )
        return self._gql_status

    @property
    def status_description(self) -> str:
        """A description of the status."""
        if hasattr(self, "_status_description"):
            return self._status_description

        self._status_description = self._extract_str_field(
            self._status_metadata, "status_description"
        )
        return self._status_description

    def _get_status_diagnostic_record(self) -> dict[str, t.Any]:
        if self._status_diagnostic_record is not None:
            return self._status_diagnostic_record

        self._status_diagnostic_record = self._status_metadata.get(
            "diagnostic_record", {}
        )
        if not isinstance(self._status_diagnostic_record, dict):
            self._status_diagnostic_record = {}
        return self._status_diagnostic_record

    @property
    def position(self) -> SummaryInputPosition | None:
        """
        The position of the input that caused the status (if applicable).

        This is vendor-specific information.

        Only notifications (see :attr:`.is_notification`) have a meaningful
        position.

        The value is :data:`None` if the server's data was missing or could not
        be interpreted.
        """
        if hasattr(self, "_position"):
            return self._position

        diag_record = self._get_status_diagnostic_record()
        self._position = SummaryInputPosition._from_metadata(
            diag_record.get("_position")
        )
        return self._position

    @property
    def raw_classification(self) -> str | None:
        """
        The raw (``str``) classification of the status.

        This is a vendor-specific classification that can be used to filter
        notifications.

        Only notifications (see :attr:`.is_notification`) have a meaningful
        classification.
        """
        if hasattr(self, "_raw_classification"):
            return self._raw_classification

        diag_record = self._get_status_diagnostic_record()
        self._raw_classification = self._extract_str_field(
            diag_record, "_classification", None
        )
        return self._raw_classification

    @property
    def classification(self) -> NotificationClassification:
        """
        Parsed version of :attr:`.raw_classification`.

        Only notifications (see :attr:`.is_notification`) have a meaningful
        classification.
        """
        if hasattr(self, "_classification"):
            return self._classification

        self._classification = _CLASSIFICATION_LOOKUP.get(
            self.raw_classification, NotificationClassification.UNKNOWN
        )
        return self._classification

    @property
    def raw_severity(self) -> str | None:
        """
        The raw (``str``) severity of the status.

        This is a vendor-specific severity that can be used to filter
        notifications.

        Only notifications (see :attr:`.is_notification`) have a meaningful
        severity.
        """
        if hasattr(self, "_raw_severity"):
            return self._raw_severity

        diag_record = self._get_status_diagnostic_record()
        self._raw_severity = self._extract_str_field(
            diag_record, "_severity", None
        )
        return self._raw_severity

    @property
    def severity(self) -> NotificationSeverity:
        """
        Parsed version of :attr:`.raw_severity`.

        Only notifications (see :attr:`.is_notification`) have a meaningful
        severity.
        """
        if hasattr(self, "_severity"):
            return self._severity

        self._severity = _SEVERITY_LOOKUP.get(
            self.raw_severity, NotificationSeverity.UNKNOWN
        )
        return self._severity

    @property
    def diagnostic_record(self) -> dict[str, t.Any]:
        """Further information about the GQLSTATUS for diagnostic purposes."""
        if hasattr(self, "_diagnostic_record"):
            return self._diagnostic_record

        self._diagnostic_record = deepcopy(
            self._get_status_diagnostic_record()
        )
        return self._diagnostic_record