USA

"""

import itertools

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

from ._dns import (

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.

"""

expired = [record for record in itertools.chain(*self.cache.values()) if record.is_expired(now)]

expired = [record for records in self.cache.values() for record in records if record.is_expired(now)]

self.async_remove_records(expired)

return expired

return None

return store.get(entry)

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

"""Gets all matching entries by details.

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

the event loop.

"""

return self._async_all_by_details(name, type_, class_)

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

def async_mark_unique_records_older_than_1s_to_expire(

self, unique_types: Set[Tuple[_str, _int, _int]], answers: Iterable[DNSRecord], now: _float

) -> None:

self._async_mark_unique_records_older_than_1s_to_expire(unique_types, answers, now)

def _async_mark_unique_records_older_than_1s_to_expire(

self, unique_types: Set[Tuple[_str, _int, _int]], answers: Iterable[DNSRecord], now: _float

) -> None:

# rfc6762#section-10.2 para 2

# Since unique is set, all old records with that name, rrtype,

# and rrclass that were received more than one second ago are declared

# invalid, and marked to expire from the cache in one second.

answers_rrset = set(answers)

for name, type_, class_ in unique_types:

for record in self._async_all_by_details(name, type_, class_):

if (now - record.created > _ONE_SECOND) and record not in answers_rrset:

for record in self.async_all_by_details(name, type_, class_):

created_float = record.created

if (now - created_float > _ONE_SECOND) and record not in answers_rrset:

# Expire in 1s

record.set_created_ttl(now, 1)