<commit_before>"""
Collection of KBucket classes storing contacts for RoutingTables.

Classes:
    KBucket -- A classic k-bucket.
    CachingKBucket -- A KBucket with a replacement cache.

Exceptions:
    FullBucketError -- Raised when adding a contact to a full KBucket.
"""

import collections
import functools
import time

from dht import constants, util


class FullBucketError(Exception):
    """Raised when attempting to add a contact in a full KBucket."""
    pass


class KBucket(collections.Sequence):
    """A simple k-bucket for a Kademlia routing table."""

    def __init__(self, range_min, range_max):
        """
        Make a new KBucket with the specified range.

        Args:
            range_min, range_max: The lower and upper limits for the
                range in the ID space covered by this KBucket, as integers.
                This is a half-open range: [range_min, range_max)
        """
        self.last_accessed = int(time.time())
        self.range_min = range_min
        self.range_max = range_max

        # This list of contacts implements an LRU-protocol:
        # Fresh or updated contacts are near the tail of the list,
        # while stale contacts are near the head.
        self._contacts = []

    # pylint: disable=no-self-argument
    # pylint: disable=not-callable
    def _touch(func):
        """
        Update the `last_accessed` attribute upon method invocation.
        """
        @functools.wraps(func)
        def updating_method(self, *args, **kwargs):
            self.last_accessed = int(time.time())
            return func(self, *args, **kwargs)
        return updating_method

    def __getitem__(self, key):
        return self._contacts[key]

    def __len__(self):
        return len(self._contacts)

    @_touch
    def add_contact(self, contact):
        """
        Add a contact to the contact list.

        The new contact is always appended to the contact list after
        removing any prior occurences of the same contact. This is the
        intended behaviour; the fresh contact may have updated add-on
        data (e.g. optimization-specific stuff).

        Args:
            contact: The contact to add, as a contact.Contact.

        Raises:
            FullBucketError: The bucket is full and the contact to add
                is not already in it.
        """
        try:
            self._contacts.remove(contact)
        except ValueError:
            pass

        if len(self._contacts) < constants.K:
            self._contacts.append(contact)
        else:
            raise FullBucketError('No space in bucket to insert contact')

    @_touch
    def get_contact(self, guid):
        """
        Return the contact with the specified guid or None if not present.

        Args:
            guid: The guid to search for, as a string or a unicode,
                in hexadecimal.

        Returns:
            A contact.Contact with the given guid or None
        """
        for contact in self._contacts:
            if contact.guid == guid:
                return contact
        return None

    @_touch<commit_msg>Assert contacts are added to proper KBucket.<commit_after>"""
Collection of KBucket classes storing contacts for RoutingTables.

Classes:
    KBucket -- A classic k-bucket.
    CachingKBucket -- A KBucket with a replacement cache.

Exceptions:
    FullBucketError -- Raised when adding a contact to a full KBucket.
"""

import collections
import functools
import time

from dht import constants, util


class FullBucketError(Exception):
    """Raised when attempting to add a contact in a full KBucket."""
    pass


class KBucket(collections.Sequence):
    """A simple k-bucket for a Kademlia routing table."""

    def __init__(self, range_min, range_max):
        """
        Make a new KBucket with the specified range.

        Args:
            range_min, range_max: The lower and upper limits for the
                range in the ID space covered by this KBucket, as integers.
                This is a half-open range: [range_min, range_max)
        """
        self.last_accessed = int(time.time())
        self.range_min = range_min
        self.range_max = range_max

        # This list of contacts implements an LRU-protocol:
        # Fresh or updated contacts are near the tail of the list,
        # while stale contacts are near the head.
        self._contacts = []

    # pylint: disable=no-self-argument
    # pylint: disable=not-callable
    def _touch(func):
        """
        Update the `last_accessed` attribute upon method invocation.
        """
        @functools.wraps(func)
        def updating_method(self, *args, **kwargs):
            self.last_accessed = int(time.time())
            return func(self, *args, **kwargs)
        return updating_method

    def __getitem__(self, key):
        return self._contacts[key]

    def __len__(self):
        return len(self._contacts)

    @_touch
    def add_contact(self, contact):
        """
        Add a contact to the contact list.

        The new contact is always appended to the contact list after
        removing any prior occurences of the same contact. This is the
        intended behaviour; the fresh contact may have updated add-on
        data (e.g. optimization-specific stuff).

        Args:
            contact: The contact to add, as a contact.Contact.

        Raises:
            FullBucketError: The bucket is full and the contact to add
                is not already in it.
        """
        assert self.contact_in_range(contact), 'Wrong KBucket.'
        try:
            self._contacts.remove(contact)
        except ValueError:
            pass

        if len(self._contacts) < constants.K:
            self._contacts.append(contact)
        else:
            raise FullBucketError('No space in bucket to insert contact')

    @_touch
    def get_contact(self, guid):
        """
        Return the contact with the specified guid or None if not present.

        Args:
            guid: The guid to search for, as a string or a unicode,
                in hexadecimal.

        Returns:
            A contact.Contact with the given guid or None
        """
        for contact in self._contacts:
            if contact.guid == guid:
                return contact
        return None

    @_touch
{'fraction_non_alphanumeric': 0.07243397121619484, 'fraction_numerical': 0.0, 'mean_word_length': 2.063953488372093, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4175160', 'n_tokens_mistral': 1695, 'n_tokens_neox': 1638, 'n_words': 639}