USA

"""

from __future__ import annotations

from heapq import heapify, heappop, heappush

from typing import Dict, Iterable, List, Optional, Set, Tuple, Union, cast

from typing import Dict, Iterable, Union, cast

from ._dns import (

DNSAddress,

def __init__(self) -> None:

self.cache: _DNSRecordCacheType = {}

self._expire_heap: List[Tuple[float, DNSRecord]] = []

self._expirations: Dict[DNSRecord, float] = {}

self._expire_heap: list[tuple[float, DNSRecord]] = []

self._expirations: dict[DNSRecord, float] = {}

self.service_cache: _DNSRecordCacheType = {}

# Functions prefixed with async_ are NOT threadsafe and must

for entry in entries:

self._async_remove(entry)

def async_expire(self, now: _float) -> List[DNSRecord]:

def async_expire(self, now: _float) -> list[DNSRecord]:

"""Purge expired entries from the cache.

This function must be run in from event loop.

if not (expire_heap_len := len(self._expire_heap)):

return []

expired: List[DNSRecord] = []

expired: list[DNSRecord] = []

# Find any expired records and add them to the to-delete list

while self._expire_heap:

when_record = self._expire_heap[0]

self.async_remove_records(expired)

return expired

def async_get_unique(self, entry: _UniqueRecordsType) -> Optional[DNSRecord]:

def async_get_unique(self, entry: _UniqueRecordsType) -> DNSRecord | None:

"""Gets a unique entry by key. Will return None if there is no

matching entry.

return None

return store.get(entry)

def async_all_by_details(self, name: _str, type_: _int, class_: _int) -> List[DNSRecord]:

def async_all_by_details(self, name: _str, type_: _int, class_: _int) -> list[DNSRecord]:

"""Gets all matching entries by details.

This function is not thread-safe and must be called from

the event loop.

"""

key = name.lower()

records = self.cache.get(key)

matches: List[DNSRecord] = []

matches: list[DNSRecord] = []

if records is None:

return matches

for record in records.values():

if type_ == record.type and class_ == record.class_:

matches.append(record)

return matches

def async_entries_with_name(self, name: str) -> List[DNSRecord]:

def async_entries_with_name(self, name: str) -> list[DNSRecord]:

"""Returns a dict of entries whose key matches the name.

This function is not threadsafe and must be called from

the event loop.

"""

return self.entries_with_name(name)

def async_entries_with_server(self, name: str) -> List[DNSRecord]:

def async_entries_with_server(self, name: str) -> list[DNSRecord]:

"""Returns a dict of entries whose key matches the server.

This function is not threadsafe and must be called from

# event loop, however they all make copies so they significantly

# inefficient.

def get(self, entry: DNSEntry) -> Optional[DNSRecord]:

def get(self, entry: DNSEntry) -> DNSRecord | None:

"""Gets an entry by key. Will return None if there is no

matching entry."""

if isinstance(entry, _UNIQUE_RECORD_TYPES):

return cached_entry

return None

def get_by_details(self, name: str, type_: _int, class_: _int) -> Optional[DNSRecord]:

def get_by_details(self, name: str, type_: _int, class_: _int) -> DNSRecord | None:

"""Gets the first matching entry by details. Returns None if no entries match.

Calling this function is not recommended as it will only

return cached_entry

return None

def get_all_by_details(self, name: str, type_: _int, class_: _int) -> List[DNSRecord]:

def get_all_by_details(self, name: str, type_: _int, class_: _int) -> list[DNSRecord]:

"""Gets all matching entries by details."""

key = name.lower()

records = self.cache.get(key)

if records is None:

return []

return [entry for entry in list(records.values()) if type_ == entry.type and class_ == entry.class_]

def entries_with_server(self, server: str) -> List[DNSRecord]:

def entries_with_server(self, server: str) -> list[DNSRecord]:

"""Returns a list of entries whose server matches the name."""

if entries := self.service_cache.get(server.lower()):

return list(entries.values())

return []

def entries_with_name(self, name: str) -> List[DNSRecord]:

def entries_with_name(self, name: str) -> list[DNSRecord]:

"""Returns a list of entries whose key matches the name."""

if entries := self.cache.get(name.lower()):

return list(entries.values())

return []

def current_entry_with_name_and_alias(self, name: str, alias: str) -> Optional[DNSRecord]:

def current_entry_with_name_and_alias(self, name: str, alias: str) -> DNSRecord | None:

now = current_time_millis()

for record in reversed(self.entries_with_name(name)):

if (

return record

return None

def names(self) -> List[str]:

def names(self) -> list[str]:

"""Return a copy of the list of current cache names."""

return list(self.cache)

def async_mark_unique_records_older_than_1s_to_expire(

self,

unique_types: Set[Tuple[_str, _int, _int]],

unique_types: set[tuple[_str, _int, _int]],

answers: Iterable[DNSRecord],

now: _float,

) -> None: