File: lru_cache.py

package info (click to toggle)
python-apptools 4.4.0-3
  • links: PTS, VCS
  • area: main
  • in suites: buster
  • size: 2,652 kB
  • sloc: python: 16,657; makefile: 77
file content (114 lines) | stat: -rw-r--r-- 3,173 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
# -*- coding: utf-8 -*-

# -----------------------------------------------------------------------------
# Copyright (c) 2015, Enthought, Inc.
# All rights reserved.
#
# This software is provided without warranty under the terms of the BSD
# license included in enthought/LICENSE.txt and may be redistributed only
# under the conditions described in the aforementioned license.  The license
# is also available online at http://www.enthought.com/licenses/BSD.txt
# Thanks for using Enthought open source!
#
# Author: Enthought, Inc.
# -----------------------------------------------------------------------------

from threading import RLock

try:
    from collections import OrderedDict
except ImportError:
    from ordereddict import OrderedDict


from traits.api import Callable, Event, HasStrictTraits, Instance, Int


class LRUCache(HasStrictTraits):
    """ A least-recently used cache.

    Items older than `size()` accesses are dropped from the cache.

    """

    size = Int

    # Called with the key and value that was dropped from the cache
    cache_drop_callback = Callable

    # This event contains the set of cached cell keys whenever it changes
    updated = Event()

    _lock = Instance(RLock, args=())

    _cache = Instance(OrderedDict)

    def __init__(self, size, **traits):
        self.size = size
        self._initialize_cache()
        super(LRUCache, self).__init__(**traits)

    def _initialize_cache(self):
        with self._lock:
            if self._cache is None:
                self._cache = OrderedDict()
            else:
                self._cache.clear()

    def _renew(self, key):
        with self._lock:
            r = self._cache.pop(key)
            self._cache[key] = r
        return r

    # -------------------------------------------------------------------------
    # LRUCache interface
    # -------------------------------------------------------------------------

    def __contains__(self, key):
        with self._lock:
            return key in self._cache

    def __len__(self):
        with self._lock:
            return len(self._cache)

    def __getitem__(self, key):
        with self._lock:
            return self._renew(key)

    def __setitem__(self, key, result):
        try:
            dropped = None
            with self._lock:
                self._cache[key] = result
                self._renew(key)
                if self.size < len(self._cache):
                    dropped = self._cache.popitem(last=False)
            if dropped and self.cache_drop_callback is not None:
                self.cache_drop_callback(*dropped)
        finally:
            self.updated = self.keys()

    def get(self, key, default=None):
        try:
            return self[key]
        except KeyError:
            return default

    def items(self):
        with self._lock:
            return self._cache.items()

    def keys(self):
        with self._lock:
            return self._cache.keys()

    def values(self):
        with self._lock:
            return self._cache.values()

    def clear(self):
        with self._lock:
            self._initialize_cache()
        self.updated = []