golden hour

/lib/python3.9/site-packages/dns

File/Folder	Size	Actions
__init__.py	1.62 KB
__pycache__	-
_asyncbackend.py	2.3 KB
_asyncio_backend.py	8.76 KB
_ddr.py	5.12 KB
_features.py	2.33 KB
_immutable_ctx.py	2.4 KB
asyncbackend.py	2.73 KB
asyncquery.py	26.22 KB
asyncresolver.py	17.43 KB
dnssec.py	39.74 KB
dnssecalgs	-
dnssectypes.py	1.76 KB
e164.py	3.88 KB
edns.py	14.91 KB
entropy.py	4.14 KB
enum.py	3.6 KB
exception.py	5.82 KB
flags.py	2.69 KB
grange.py	2.1 KB
immutable.py	1.97 KB
inet.py	5.64 KB
ipv4.py	2.49 KB
ipv6.py	6.45 KB
message.py	64.45 KB
name.py	41.67 KB
namedict.py	3.91 KB
nameserver.py	9.68 KB
node.py	12.37 KB
opcode.py	2.67 KB
query.py	53.55 KB
quic	-
rcode.py	4.06 KB
rdata.py	28.77 KB
rdataclass.py	2.91 KB
rdataset.py	16.36 KB
rdatatype.py	7.17 KB
rdtypes	-
renderer.py	10.99 KB
resolver.py	71.83 KB
reversename.py	3.74 KB
rrset.py	8.96 KB
serial.py	3.52 KB
set.py	8.88 KB
tokenizer.py	23.03 KB
transaction.py	22.11 KB
tsig.py	11.15 KB
tsigkeyring.py	2.57 KB
ttl.py	2.91 KB
update.py	11.96 KB
version.py	1.88 KB
versioned.py	11.49 KB
win32util.py	8.89 KB
wire.py	2.76 KB
xfr.py	12.96 KB
zone.py	50.87 KB
zonefile.py	27.27 KB
zonetypes.py	690 B

Edit: versioned.py

# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license

"""DNS Versioned Zones."""

import collections
import threading
from typing import Callable, Deque, Optional, Set, Union

import dns.exception
import dns.immutable
import dns.name
import dns.node
import dns.rdataclass
import dns.rdataset
import dns.rdatatype
import dns.rdtypes.ANY.SOA
import dns.zone

class UseTransaction(dns.exception.DNSException):
    """To alter a versioned zone, use a transaction."""

# Backwards compatibility
Node = dns.zone.VersionedNode
ImmutableNode = dns.zone.ImmutableVersionedNode
Version = dns.zone.Version
WritableVersion = dns.zone.WritableVersion
ImmutableVersion = dns.zone.ImmutableVersion
Transaction = dns.zone.Transaction

class Zone(dns.zone.Zone):  # lgtm[py/missing-equals]
    __slots__ = [
        "_versions",
        "_versions_lock",
        "_write_txn",
        "_write_waiters",
        "_write_event",
        "_pruning_policy",
        "_readers",
    ]

node_factory = Node

def __init__(
        self,
        origin: Optional[Union[dns.name.Name, str]],
        rdclass: dns.rdataclass.RdataClass = dns.rdataclass.IN,
        relativize: bool = True,
        pruning_policy: Optional[Callable[["Zone", Version], Optional[bool]]] = None,
    ):
        """Initialize a versioned zone object.

*origin* is the origin of the zone.  It may be a ``dns.name.Name``,
        a ``str``, or ``None``.  If ``None``, then the zone's origin will
        be set by the first ``$ORIGIN`` line in a zone file.

*rdclass*, an ``int``, the zone's rdata class; the default is class IN.

*relativize*, a ``bool``, determine's whether domain names are
        relativized to the zone's origin.  The default is ``True``.

*pruning policy*, a function taking a ``Zone`` and a ``Version`` and returning
        a ``bool``, or ``None``.  Should the version be pruned?  If ``None``,
        the default policy, which retains one version is used.
        """
        super().__init__(origin, rdclass, relativize)
        self._versions: Deque[Version] = collections.deque()
        self._version_lock = threading.Lock()
        if pruning_policy is None:
            self._pruning_policy = self._default_pruning_policy
        else:
            self._pruning_policy = pruning_policy
        self._write_txn: Optional[Transaction] = None
        self._write_event: Optional[threading.Event] = None
        self._write_waiters: Deque[threading.Event] = collections.deque()
        self._readers: Set[Transaction] = set()
        self._commit_version_unlocked(
            None, WritableVersion(self, replacement=True), origin
        )

def reader(
        self, id: Optional[int] = None, serial: Optional[int] = None
    ) -> Transaction:  # pylint: disable=arguments-differ
        if id is not None and serial is not None:
            raise ValueError("cannot specify both id and serial")
        with self._version_lock:
            if id is not None:
                version = None
                for v in reversed(self._versions):
                    if v.id == id:
                        version = v
                        break
                if version is None:
                    raise KeyError("version not found")
            elif serial is not None:
                if self.relativize:
                    oname = dns.name.empty
                else:
                    assert self.origin is not None
                    oname = self.origin
                version = None
                for v in reversed(self._versions):
                    n = v.nodes.get(oname)
                    if n:
                        rds = n.get_rdataset(self.rdclass, dns.rdatatype.SOA)
                        if rds and rds[0].serial == serial:
                            version = v
                            break
                if version is None:
                    raise KeyError("serial not found")
            else:
                version = self._versions[-1]
            txn = Transaction(self, False, version)
            self._readers.add(txn)
            return txn

def writer(self, replacement: bool = False) -> Transaction:
        event = None
        while True:
            with self._version_lock:
                # Checking event == self._write_event ensures that either
                # no one was waiting before we got lucky and found no write
                # txn, or we were the one who was waiting and got woken up.
                # This prevents "taking cuts" when creating a write txn.
                if self._write_txn is None and event == self._write_event:
                    # Creating the transaction defers version setup
                    # (i.e.  copying the nodes dictionary) until we
                    # give up the lock, so that we hold the lock as
                    # short a time as possible.  This is why we call
                    # _setup_version() below.
                    self._write_txn = Transaction(
                        self, replacement, make_immutable=True
                    )
                    # give up our exclusive right to make a Transaction
                    self._write_event = None
                    break
                # Someone else is writing already, so we will have to
                # wait, but we want to do the actual wait outside the
                # lock.
                event = threading.Event()
                self._write_waiters.append(event)
            # wait (note we gave up the lock!)
            #
            # We only wake one sleeper at a time, so it's important
            # that no event waiter can exit this method (e.g. via
            # cancellation) without returning a transaction or waking
            # someone else up.
            #
            # This is not a problem with Threading module threads as
            # they cannot be canceled, but could be an issue with trio
            # tasks when we do the async version of writer().
            # I.e. we'd need to do something like:
            #
            # try:
            #     event.wait()
            # except trio.Cancelled:
            #     with self._version_lock:
            #         self._maybe_wakeup_one_waiter_unlocked()
            #     raise
            #
            event.wait()
        # Do the deferred version setup.
        self._write_txn._setup_version()
        return self._write_txn

def _maybe_wakeup_one_waiter_unlocked(self):
        if len(self._write_waiters) > 0:
            self._write_event = self._write_waiters.popleft()
            self._write_event.set()

# pylint: disable=unused-argument
    def _default_pruning_policy(self, zone, version):
        return True

# pylint: enable=unused-argument

def _prune_versions_unlocked(self):
        assert len(self._versions) > 0
        # Don't ever prune a version greater than or equal to one that
        # a reader has open.  This pins versions in memory while the
        # reader is open, and importantly lets the reader open a txn on
        # a successor version (e.g. if generating an IXFR).
        #
        # Note our definition of least_kept also ensures we do not try to
        # delete the greatest version.
        if len(self._readers) > 0:
            least_kept = min(txn.version.id for txn in self._readers)
        else:
            least_kept = self._versions[-1].id
        while self._versions[0].id < least_kept and self._pruning_policy(
            self, self._versions[0]
        ):
            self._versions.popleft()

def set_max_versions(self, max_versions: Optional[int]) -> None:
        """Set a pruning policy that retains up to the specified number
        of versions
        """
        if max_versions is not None and max_versions < 1:
            raise ValueError("max versions must be at least 1")
        if max_versions is None:

def policy(zone, _):  # pylint: disable=unused-argument
                return False

else:

def policy(zone, _):
                return len(zone._versions) > max_versions

self.set_pruning_policy(policy)

def set_pruning_policy(
        self, policy: Optional[Callable[["Zone", Version], Optional[bool]]]
    ) -> None:
        """Set the pruning policy for the zone.

The *policy* function takes a `Version` and returns `True` if
        the version should be pruned, and `False` otherwise.  `None`
        may also be specified for policy, in which case the default policy
        is used.

Pruning checking proceeds from the least version and the first
        time the function returns `False`, the checking stops.  I.e. the
        retained versions are always a consecutive sequence.
        """
        if policy is None:
            policy = self._default_pruning_policy
        with self._version_lock:
            self._pruning_policy = policy
            self._prune_versions_unlocked()

def _end_read(self, txn):
        with self._version_lock:
            self._readers.remove(txn)
            self._prune_versions_unlocked()

def _end_write_unlocked(self, txn):
        assert self._write_txn == txn
        self._write_txn = None
        self._maybe_wakeup_one_waiter_unlocked()

def _end_write(self, txn):
        with self._version_lock:
            self._end_write_unlocked(txn)

def _commit_version_unlocked(self, txn, version, origin):
        self._versions.append(version)
        self._prune_versions_unlocked()
        self.nodes = version.nodes
        if self.origin is None:
            self.origin = origin
        # txn can be None in __init__ when we make the empty version.
        if txn is not None:
            self._end_write_unlocked(txn)

def _commit_version(self, txn, version, origin):
        with self._version_lock:
            self._commit_version_unlocked(txn, version, origin)

def _get_next_version_id(self):
        if len(self._versions) > 0:
            id = self._versions[-1].id + 1
        else:
            id = 1
        return id

def find_node(
        self, name: Union[dns.name.Name, str], create: bool = False
    ) -> dns.node.Node:
        if create:
            raise UseTransaction
        return super().find_node(name)

def delete_node(self, name: Union[dns.name.Name, str]) -> None:
        raise UseTransaction

def find_rdataset(
        self,
        name: Union[dns.name.Name, str],
        rdtype: Union[dns.rdatatype.RdataType, str],
        covers: Union[dns.rdatatype.RdataType, str] = dns.rdatatype.NONE,
        create: bool = False,
    ) -> dns.rdataset.Rdataset:
        if create:
            raise UseTransaction
        rdataset = super().find_rdataset(name, rdtype, covers)
        return dns.rdataset.ImmutableRdataset(rdataset)

def get_rdataset(
        self,
        name: Union[dns.name.Name, str],
        rdtype: Union[dns.rdatatype.RdataType, str],
        covers: Union[dns.rdatatype.RdataType, str] = dns.rdatatype.NONE,
        create: bool = False,
    ) -> Optional[dns.rdataset.Rdataset]:
        if create:
            raise UseTransaction
        rdataset = super().get_rdataset(name, rdtype, covers)
        if rdataset is not None:
            return dns.rdataset.ImmutableRdataset(rdataset)
        else:
            return None

def delete_rdataset(
        self,
        name: Union[dns.name.Name, str],
        rdtype: Union[dns.rdatatype.RdataType, str],
        covers: Union[dns.rdatatype.RdataType, str] = dns.rdatatype.NONE,
    ) -> None:
        raise UseTransaction

def replace_rdataset(
        self, name: Union[dns.name.Name, str], replacement: dns.rdataset.Rdataset
    ) -> None:
        raise UseTransaction