0
  $   ˆl–Ým_Jå0–54 ^¸Ëw%1hßÂ„œeÏÁÀ¿ ?÷     # Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license

import collections
from typing import Any, Callable, Iterator, List, Optional, Tuple, Union

import dns.exception
import dns.name
import dns.node
import dns.rdataclass
import dns.rdataset
import dns.rdatatype
import dns.rrset
import dns.serial
import dns.ttl


class TransactionManager:
    def reader(self) -> "Transaction":
        """Begin a read-only transaction."""
        raise NotImplementedError  # pragma: no cover

    def writer(self, replacement: bool = False) -> "Transaction":
        """Begin a writable transaction.

        *replacement*, a ``bool``.  If `True`, the content of the
        transaction completely replaces any prior content.  If False,
        the default, then the content of the transaction updates the
        existing content.
        """
        raise NotImplementedError  # pragma: no cover

    def origin_information(
        self,
    ) -> Tuple[Optional[dns.name.Name], bool, Optional[dns.name.Name]]:
        """Returns a tuple

            (absolute_origin, relativize, effective_origin)

        giving the absolute name of the default origin for any
        relative domain names, the "effective origin", and whether
        names should be relativized.  The "effective origin" is the
        absolute origin if relativize is False, and the empty name if
        relativize is true.  (The effective origin is provided even
        though it can be computed from the absolute_origin and
        relativize setting because it avoids a lot of code
        duplication.)

        If the returned names are `None`, then no origin information is
        available.

        This information is used by code working with transactions to
        allow it to coordinate relativization.  The transaction code
        itself takes what it gets (i.e. does not change name
        relativity).

        """
        raise NotImplementedError  # pragma: no cover

    def get_class(self) -> dns.rdataclass.RdataClass:
        """The class of the transaction manager."""
        raise NotImplementedError  # pragma: no cover

    def from_wire_origin(self) -> Optional[dns.name.Name]:
        """Origin to use in from_wire() calls."""
        (absolute_origin, relativize, _) = self.origin_information()
        if relativize:
            return absolute_origin
        else:
            return None


class DeleteNotExact(dns.exception.DNSException):
    """Existing data did not match data specified by an exact delete."""


class ReadOnly(dns.exception.DNSException):
    """Tried to write to a read-only transaction."""


class AlreadyEnded(dns.exception.DNSException):
    """Tried to use an already-ended transaction."""


def _ensure_immutable_rdataset(rdataset):
    if rdataset is None or isinstance(rdataset, dns.rdataset.ImmutableRdataset):
        return rdataset
    return dns.rdataset.ImmutableRdataset(rdataset)


def _ensure_immutable_node(node):
    if node is None or node.is_immutable():
        return node
    return dns.node.ImmutableNode(node)


CheckPutRdatasetType = Callable[
    ["Transaction", dns.name.Name, dns.rdataset.Rdataset], None
]
CheckDeleteRdatasetType = Callable[
    ["Transaction", dns.name.Name, dns.rdatatype.RdataType, dns.rdatatype.RdataType],
    None,
]
CheckDeleteNameType = Callable[["Transaction", dns.name.Name], None]


class Transaction:
    def __init__(
        self,
        manager: TransactionManager,
        replacement: bool = False,
        read_only: bool = False,
    ):
        self.manager = manager
        self.replacement = replacement
        self.read_only = read_only
        self._ended = False
        self._check_put_rdataset: List[CheckPutRdatasetType] = []
        self._check_delete_rdataset: List[CheckDeleteRdatasetType] = []
        self._check_delete_name: List[CheckDeleteNameType] = []

    #
    # This is the high level API
    #
    # Note that we currently use non-immutable types in the return type signature to
    # avoid covariance probl