"""Core disk and file backed cache API.

"""

import codecs

import contextlib as cl

import errno

import functools as ft

import io

import os

import os.path as op

import sqlite3

import struct

import sys

import threading

import time

import warnings

import zlib

if sys.hexversion < 0x03000000:

import cPickle as pickle

# ISSUE #25 Fix for http://bugs.python.org/issue10211

from cStringIO import StringIO as BytesIO

TextType = unicode

BytesType = str

INT_TYPES = int, long

range = xrange # pylint: disable=redefined-builtin,invalid-name

io_open = io.open # pylint: disable=invalid-name

else:

import pickle

from io import BytesIO # pylint: disable=ungrouped-imports

TextType = str

BytesType = bytes

INT_TYPES = int,

io_open = open # pylint: disable=invalid-name

try:

WindowsError

except NameError:

class WindowsError(Exception):

"Windows error place-holder on platforms without support."

pass

class Constant(tuple):

"Pretty display of immutable constant."

def __new__(cls, name):

return tuple.__new__(cls, (name,))

def __repr__(self):

return self[0]

DBNAME = 'cache.db'

ENOVAL = Constant('ENOVAL')

MODE_NONE = 0

MODE_RAW = 1

MODE_BINARY = 2

MODE_TEXT = 3

MODE_PICKLE = 4

DEFAULT_SETTINGS = {

u'statistics': 0, # False

u'tag_index': 0, # False

u'eviction_policy': u'least-recently-stored',

u'size_limit': 2 ** 30, # 1gb

u'cull_limit': 10,

u'sqlite_cache_size': 2 ** 13, # 8,192 pages

u'sqlite_journal_mode': u'WAL',

u'sqlite_mmap_size': 2 ** 26, # 64mb

u'sqlite_synchronous': u'NORMAL',

u'disk_min_file_size': 2 ** 15, # 32kb

u'disk_pickle_protocol': pickle.HIGHEST_PROTOCOL,

}

METADATA = {

u'count': 0,

u'size': 0,

u'hits': 0,

u'misses': 0,

}

EVICTION_POLICY = {

'none': {

'init': None,

'get': None,

'cull': None,

},

'least-recently-stored': {

'init': (

'CREATE INDEX IF NOT EXISTS Cache_store_time ON'

' Cache (store_time)'

),

'get': None,

'cull': 'SELECT %s FROM Cache ORDER BY store_time LIMIT ?',

},

'least-recently-used': {

'init': (

'CREATE INDEX IF NOT EXISTS Cache_access_time ON'

' Cache (access_time)'

),

'get': 'access_time = ((julianday("now") - 2440587.5) * 86400.0)',

'cull': 'SELECT %s FROM Cache ORDER BY access_time LIMIT ?',

},

'least-frequently-used': {

'init': (

'CREATE INDEX IF NOT EXISTS Cache_access_count ON'

' Cache (access_count)'

),

'get': 'access_count = access_count + 1',

'cull': 'SELECT %s FROM Cache ORDER BY access_count LIMIT ?',

},

}

class Disk(object):

"Cache key and value serialization for SQLite database and files."

def __init__(self, directory, min_file_size=0, pickle_protocol=0):

"""Initialize disk instance.

:param str directory: directory path

:param int min_file_size: minimum size for file use

:param int pickle_protocol: pickle protocol for serialization

"""

self._directory = directory

self.min_file_size = min_file_size

self.pickle_protocol = pickle_protocol

def hash(self, key):

"""Compute portable hash for `key`.

:param key: key to hash

:return: hash value

"""

mask = 0xFFFFFFFF

disk_key, _ = self.put(key)

type_disk_key = type(disk_key)

if type_disk_key is sqlite3.Binary:

return zlib.adler32(disk_key) & mask

elif type_disk_key is TextType:

return zlib.adler32(disk_key.encode('utf-8')) & mask # pylint: disable=no-member

elif type_disk_key in INT_TYPES:

return disk_key % mask

else:

assert type_disk_key is float

return zlib.adler32(struct.pack('!d', disk_key)) & mask

def put(self, key):

"""Convert `key` to fields key and raw for Cache table.

:param key: key to convert

:return: (database key, raw boolean) pair

"""

# pylint: disable=bad-continuation,unidiomatic-typecheck

type_key = type(key)

if type_key is BytesType:

return sqlite3.Binary(key), True

elif ((type_key is TextType)

or (type_key in INT_TYPES

and -9223372036854775808 <= key <= 9223372036854775807)

or (type_key is float)):

return key, True

else:

result = pickle.dumps(key, protocol=self.pickle_protocol)

return sqlite3.Binary(result), False

def get(self, key, raw):

"""Convert fields `key` and `raw` from Cache table to key.

:param key: database key to convert

:param bool raw: flag indicating raw database storage

:return: corresponding Python key

"""

# pylint: disable=no-self-use,unidiomatic-typecheck

if raw:

return BytesType(key) if type(key) is sqlite3.Binary else key

else:

return pickle.load(BytesIO(key))

def store(self, value, read):

"""Convert `value` to fields size, mode, filename, and value for Cache

table.

:param value: value to convert

:param bool read: True when value is file-like object

:return: (size, mode, filename, value) tuple for Cache table

"""

# pylint: disable=unidiomatic-typecheck

type_value = type(value)

min_file_size = self.min_file_size

if ((type_value is TextType and len(value) < min_file_size)

or (type_value in INT_TYPES

and -9223372036854775808 <= value <= 9223372036854775807)

or (type_value is float)):

return 0, MODE_RAW, None, value

elif type_value is BytesType:

if len(value) < min_file_size:

return 0, MODE_RAW, None, sqlite3.Binary(value)

else:

filename, full_path = self.filename()

with open(full_path, 'wb') as writer:

writer.write(value)

return len(value), MODE_BINARY, filename, None

elif type_value is TextType:

filename, full_path = self.filename()

with io_open(full_path, 'w', encoding='UTF-8') as writer:

writer.write(value)

size = op.getsize(full_path)

return size, MODE_TEXT, filename, None

elif read:

size = 0

reader = ft.partial(value.read, 2 ** 22)

filename, full_path = self.filename()

with open(full_path, 'wb') as writer:

for chunk in iter(reader, b''):

size += len(chunk)

writer.write(chunk)

return size, MODE_BINARY, filename, None

else:

result = pickle.dumps(value, protocol=self.pickle_protocol)

if len(result) < min_file_size:

return 0, MODE_PICKLE, None, sqlite3.Binary(result)

else:

filename, full_path = self.filename()

with open(full_path, 'wb') as writer:

writer.write(result)

return len(result), MODE_PICKLE, filename, None

def fetch(self, mode, filename, value, read):

"""Convert fields `mode`, `filename`, and `value` from Cache table to

value.

:param int mode: value mode raw, binary, text, or pickle

:param str filename: filename of corresponding value

:param value: database value

:param bool read: when True, return an open file handle

:return: corresponding Python value

"""

# pylint: disable=no-self-use,unidiomatic-typecheck

if mode == MODE_RAW:

return BytesType(value) if type(value) is sqlite3.Binary else value

elif mode == MODE_BINARY:

if read:

return open(op.join(self._directory, filename), 'rb')

else:

with open(op.join(self._directory, filename), 'rb') as reader:

return reader.read()

elif mode == MODE_TEXT:

full_path = op.join(self._directory, filename)

with io_open(full_path, 'r', encoding='UTF-8') as reader:

return reader.read()

elif mode == MODE_PICKLE:

if value is None:

with open(op.join(self._directory, filename), 'rb') as reader:

return pickle.load(reader)

else:

return pickle.load(BytesIO(value))

def filename(self):

"""Return filename and full-path tuple for file storage.

Filename will be a randomly generated 28 character hexadecimal string

with ".val" suffixed. Two levels of sub-directories will be used to

reduce the size of directories. On older filesystems, lookups in

directories with many files are slow.

"""

hex_name = codecs.encode(os.urandom(16), 'hex').decode('utf-8')

sub_dir = op.join(hex_name[:2], hex_name[2:4])

name = hex_name[4:] + '.val'

directory = op.join(self._directory, sub_dir)

try:

os.makedirs(directory)

except OSError as error:

if error.errno != errno.EEXIST:

raise

filename = op.join(sub_dir, name)

full_path = op.join(self._directory, filename)

return filename, full_path

def remove(self, filename):

"""Remove a file given by `filename`.

This method is cross-thread and cross-process safe. If an "error no

entry" occurs, it is suppressed.

:param str filename: relative path to file

"""

full_path = op.join(self._directory, filename)

try:

os.remove(full_path)

except WindowsError:

pass

except OSError as error:

if error.errno != errno.ENOENT:

# ENOENT may occur if two caches attempt to delete the same

# file at the same time.

raise

class Timeout(Exception):

"Database timeout expired."

pass

class UnknownFileWarning(UserWarning):

"Warning used by Cache.check for unknown files."

pass

class EmptyDirWarning(UserWarning):

"Warning used by Cache.check for empty directories."

pass

class Cache(object):

"Disk and file backed cache."

# pylint: disable=bad-continuation

def __init__(self, directory, timeout=60, disk=Disk, **settings):

"""Initialize cache instance.

:param str directory: cache directory

:param float timeout: SQLite connection timeout

:param disk: Disk type or subclass for serialization

:param settings: any of DEFAULT_SETTINGS

"""

try:

assert issubclass(disk, Disk)

except (TypeError, AssertionError):

raise ValueError('disk must subclass diskcache.Disk')

self._directory = directory

self._timeout = 60 # Use 1 minute timeout for initialization.

self._local = threading.local()

if not op.isdir(directory):

try:

os.makedirs(directory, 0o755)

except OSError as error:

if error.errno != errno.EEXIST:

raise EnvironmentError(

error.errno,

'Cache directory "%s" does not exist'

' and could not be created' % self._directory

)

sql = self._sql

# Setup Settings table.

sql('CREATE TABLE IF NOT EXISTS Settings ('

' key TEXT NOT NULL UNIQUE,'

' value)'

)

current_settings = dict(sql(

'SELECT key, value FROM Settings'

).fetchall())

sets = DEFAULT_SETTINGS.copy()

sets.update(current_settings)

sets.update(settings)

for key in METADATA:

sets.pop(key, None)

# Setup Disk object (must happen after settings initialized).

kwargs = {

key[5:]: value for key, value in sets.items()

if key.startswith('disk_')

}

self._disk = disk(directory, **kwargs)

# Set cached attributes: updates settings and sets pragmas.

for key, value in sets.items():

query = 'INSERT OR REPLACE INTO Settings VALUES (?, ?)'

sql(query, (key, value))

self.reset(key, value)

for key, value in METADATA.items():

query = 'INSERT OR IGNORE INTO Settings VALUES (?, ?)'

sql(query, (key, value))

self.reset(key)

(self._page_size,), = sql('PRAGMA page_size').fetchall()

# Setup Cache table.

sql('CREATE TABLE IF NOT EXISTS Cache ('

' rowid INTEGER PRIMARY KEY,'

' key BLOB,'

' raw INTEGER,'

' version INTEGER DEFAULT 0,'

' store_time REAL,'

' expire_time REAL,'

' access_time REAL,'

' access_count INTEGER DEFAULT 0,'

' tag BLOB,'

' size INTEGER DEFAULT 0,'

' mode INTEGER DEFAULT 0,'

' filename TEXT,'

' value BLOB)'

)

sql('CREATE UNIQUE INDEX IF NOT EXISTS Cache_key_raw ON'

' Cache(key, raw)'

)

sql('CREATE INDEX IF NOT EXISTS Cache_expire_time ON'

' Cache (expire_time)'

)

query = EVICTION_POLICY[self.eviction_policy]['init']

if query is not None:

sql(query)

# Use triggers to keep Metadata updated.

sql('CREATE TRIGGER IF NOT EXISTS Settings_count_insert'

' AFTER INSERT ON Cache FOR EACH ROW BEGIN'

' UPDATE Settings SET value = value + 1'

' WHERE key = "count"; END'

)

sql('CREATE TRIGGER IF NOT EXISTS Settings_count_delete'

' AFTER DELETE ON Cache FOR EACH ROW BEGIN'

' UPDATE Settings SET value = value - 1'

' WHERE key = "count"; END'

)

sql('CREATE TRIGGER IF NOT EXISTS Settings_size_insert'

' AFTER INSERT ON Cache FOR EACH ROW BEGIN'

' UPDATE Settings SET value = value + NEW.size'

' WHERE key = "size"; END'

)

sql('CREATE TRIGGER IF NOT EXISTS Settings_size_update'

' AFTER UPDATE ON Cache FOR EACH ROW BEGIN'

' UPDATE Settings'

' SET value = value + NEW.size - OLD.size'

' WHERE key = "size"; END'

)

sql('CREATE TRIGGER IF NOT EXISTS Settings_size_delete'

' AFTER DELETE ON Cache FOR EACH ROW BEGIN'

' UPDATE Settings SET value = value - OLD.size'

' WHERE key = "size"; END'

)

# Create tag index if requested.

if self.tag_index: # pylint: disable=no-member

self.create_tag_index()

else:

self.drop_tag_index()

# Close and re-open database connection with given timeout.

self.close()

self._timeout = timeout

assert self._sql

@property

def directory(self):

"""Cache directory."""

return self._directory

@property

def timeout(self):

"""SQLite connection timeout value in seconds."""

return self._timeout

@property

def disk(self):

"""Disk used for serialization."""

return self._disk

@property

def _sql(self):

con = getattr(self._local, 'con', None)

if con is None:

con = self._local.con = sqlite3.connect(

op.join(self._directory, DBNAME),

timeout=self._timeout,

isolation_level=None,

)

return con.execute

@cl.contextmanager

def _transact(self, filename=None):

sql = self._sql

filenames = []

_disk_remove = self._disk.remove

try:

sql('BEGIN IMMEDIATE')

except sqlite3.OperationalError:

if filename is not None:

_disk_remove(filename)

raise Timeout

try:

yield sql, filenames.append

except BaseException:

sql('ROLLBACK')

raise

else:

sql('COMMIT')

for filename in filenames:

if filename is not None:

_disk_remove(filename)

def set(self, key, value, expire=None, read=False, tag=None):

"""Set `key` and `value` item in cache.

When `read` is `True`, `value` should be a file-like object opened

for reading in binary mode.

:param key: key for item

:param value: value for item

:param float expire: seconds until item expires

(default None, no expiry)

:param bool read: read value as bytes from file (default False)

:param str tag: text to associate with key (default None)

:return: True if item was set

:raises Timeout: if database timeout expires

"""

now = time.time()

db_key, raw = self._disk.put(key)

expire_time = None if expire is None else now + expire

size, mode, filename, db_value = self._disk.store(value, read)

columns = (expire_time, tag, size, mode, filename, db_value)

# The order of SELECT, UPDATE, and INSERT is important below.

#

# Typical cache usage pattern is:

#

# value = cache.get(key)

# if value is None:

# value = expensive_calculation()

# cache.set(key, value)

#

# Cache.get does not evict expired keys to avoid writes during lookups.

# Commonly used/expired keys will therefore remain in the cache making

# an UPDATE the preferred path.

#

# The alternative is to assume the key is not present by first trying

# to INSERT and then handling the IntegrityError that occurs from

# violating the UNIQUE constraint. This optimistic approach was

# rejected based on the common cache usage pattern.

#

# INSERT OR REPLACE aka UPSERT is not used because the old filename may

# need cleanup.

with self._transact(filename) as (sql, cleanup):

rows = sql(

'SELECT rowid, filename FROM Cache'

' WHERE key = ? AND raw = ?',

(db_key, raw),

).fetchall()

if rows:

(rowid, old_filename), = rows

cleanup(old_filename)

self._row_update(rowid, now, columns)

else:

self._row_insert(db_key, raw, now, columns)

self._cull(now, sql, cleanup)

return True

__setitem__ = set

def _row_update(self, rowid, now, columns):

sql = self._sql

expire_time, tag, size, mode, filename, value = columns

sql('UPDATE Cache SET'

' version = ?,'

' store_time = ?,'

' expire_time = ?,'

' access_time = ?,'

' access_count = ?,'

' tag = ?,'

' size = ?,'

' mode = ?,'

' filename = ?,'

' value = ?'

' WHERE rowid = ?', (

0, # version

now, # store_time

expire_time,

now, # access_time

0, # access_count

tag,

size,

mode,

filename,

value,

rowid,

),

)

def _row_insert(self, key, raw, now, columns):

sql = self._sql

expire_time, tag, size, mode, filename, value = columns

sql('INSERT INTO Cache('

' key, raw, version, store_time, expire_time, access_time,'

' access_count, tag, size, mode, filename, value'

') VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', (

key,

raw,

0, # version

now, # store_time

expire_time,

now, # access_time

0, # access_count

tag,

size,

mode,

filename,

value,

),

)

def _cull(self, now, sql, cleanup):

cull_limit = self.cull_limit

if cull_limit == 0:

return

# Evict expired keys.

select_expired_template = (

'SELECT %s FROM Cache'

' WHERE expire_time IS NOT NULL AND expire_time < ?'

' ORDER BY expire_time LIMIT ?'

)

select_expired = select_expired_template % 'filename'

rows = sql(select_expired, (now, cull_limit)).fetchall()

if rows:

delete_expired = (

'DELETE FROM Cache WHERE rowid IN (%s)'

% (select_expired_template % 'rowid')

)

sql(delete_expired, (now, cull_limit))

for filename, in rows:

cleanup(filename)

cull_limit -= len(rows)

if cull_limit == 0:

return

# Evict keys by policy.

select_policy_template = EVICTION_POLICY[self.eviction_policy]['cull']

if select_policy_template is None or self.volume() < self.size_limit:

return

select_policy = select_policy_template % 'filename'

rows = sql(select_policy, (cull_limit,)).fetchall()

if rows:

delete_policy = (

'DELETE FROM Cache WHERE rowid IN (%s)'

% (select_policy_template % 'rowid')

)

sql(delete_policy, (cull_limit,))

for filename, in rows:

cleanup(filename)

def add(self, key, value, expire=None, read=False, tag=None):

"""Add `key` and `value` item to cache.

Similar to `set`, but only add to cache if key not present.

Operation is atomic. Only one concurrent add operation for a given key

will succeed.

When `read` is `True`, `value` should be a file-like object opened

for reading in binary mode.

:param key: key for item

:param value: value for item

:param float expire: seconds until the key expires

(default None, no expiry)

:param bool read: read value as bytes from file (default False)

:param str tag: text to associate with key (default None)

:return: True if item was added

:raises Timeout: if database timeout expires

"""

now = time.time()

db_key, raw = self._disk.put(key)

expire_time = None if expire is None else now + expire

size, mode, filename, db_value = self._disk.store(value, read)

columns = (expire_time, tag, size, mode, filename, db_value)

with self._transact(filename) as (sql, cleanup):

rows = sql(

'SELECT rowid, filename, expire_time FROM Cache'

' WHERE key = ? AND raw = ?',

(db_key, raw),

).fetchall()

if rows:

(rowid, old_filename, old_expire_time), = rows

if old_expire_time is None or old_expire_time > now:

cleanup(filename)

return False

cleanup(old_filename)

self._row_update(rowid, now, columns)

else:

self._row_insert(db_key, raw, now, columns)

self._cull(now, sql, cleanup)

return True

def incr(self, key, delta=1, default=0):

"""Increment value by delta for item with key.

If key is missing and default is None then raise KeyError. Else if key

is missing and default is not None then use default for value.

Operation is atomic. All concurrent increment operations will be

counted individually.

Assumes value may be stored in a SQLite column. Most builds that target

machines with 64-bit pointer widths will support 64-bit signed

integers.

:param key: key for item

:param int delta: amount to increment (default 1)

:param int default: value if key is missing (default None)

:return: new value for item

:raises KeyError: if key is not found and default is None

:raises Timeout: if database timeout expires

"""

now = time.time()

db_key, raw = self._disk.put(key)

select = (

'SELECT rowid, expire_time, filename, value FROM Cache'

' WHERE key = ? AND raw = ?'

)

with self._transact() as (sql, cleanup):

rows = sql(select, (db_key, raw)).fetchall()

if not rows:

if default is None:

raise KeyError(key)

value = default + delta

columns = (None, None) + self._disk.store(value, False)

self._row_insert(db_key, raw, now, columns)

self._cull(now, sql, cleanup)

return value

(rowid, expire_time, filename, value), = rows

if expire_time is not None and expire_time < now:

if default is None:

raise KeyError(key)

value = default + delta

columns = (None, None) + self._disk.store(value, False)

self._row_update(rowid, now, columns)

self._cull(now, sql, cleanup)

cleanup(filename)

return value

value += delta

columns = 'store_time = ?, value = ?'

update_column = EVICTION_POLICY[self.eviction_policy]['get']

columns += '' if update_column is None else ', ' + update_column

update = 'UPDATE Cache SET %s WHERE rowid = ?' % columns

sql(update, (now, value, rowid))

return value

def decr(self, key, delta=1, default=0):

"""Decrement value by delta for item with key.

If key is missing and default is None then raise KeyError. Else if key

is missing and default is not None then use default for value.

Operation is atomic. All concurrent decrement operations will be

counted individually.

Unlike Memcached, negative values are supported. Value may be

decremented below zero.

Assumes value may be stored in a SQLite column. Most builds that target

machines with 64-bit pointer widths will support 64-bit signed

integers.

:param key: key for item

:param int delta: amount to decrement (default 1)

:param int default: value if key is missing (default 0)

:return: new value for item

:raises KeyError: if key is not found and default is None

:raises Timeout: if database timeout expires

"""

return self.incr(key, -delta, default)

def get(self, key, default=None, read=False, expire_time=False, tag=False):

"""Retrieve value from cache. If `key` is missing, return `default`.

:param key: key for item

:param default: value to return if key is missing (default None)

:param bool read: if True, return file handle to value

(default False)

:param bool expire_time: if True, return expire_time in tuple

(default False)

:param bool tag: if True, return tag in tuple (default False)

:return: value for item or default if key not found

:raises Timeout: if database timeout expires

"""

db_key, raw = self._disk.put(key)

update_column = EVICTION_POLICY[self.eviction_policy]['get']

update = 'UPDATE Cache SET %s WHERE rowid = ?'

select = (

'SELECT rowid, expire_time, tag, mode, filename, value'

' FROM Cache WHERE key = ? AND raw = ?'

' AND (expire_time IS NULL OR expire_time > ?)'

)

if expire_time and tag:

default = (default, None, None)

elif expire_time or tag:

default = (default, None)

if not self.statistics and update_column is None:

# Fast path, no transaction necessary.

rows = self._sql(select, (db_key, raw, time.time())).fetchall()

if not rows:

return default

(rowid, db_expire_time, db_tag, mode, filename, db_value), = rows

try:

value = self._disk.fetch(mode, filename, db_value, read)

except IOError as error:

if error.errno == errno.ENOENT:

# Key was deleted before we could retrieve result.

return default

else:

raise

else: # Slow path, transaction required.

cache_hit = (

'UPDATE Settings SET value = value + 1 WHERE key = "hits"'

)

cache_miss = (

'UPDATE Settings SET value = value + 1 WHERE key = "misses"'

)

with self._transact() as (sql, _):

rows = sql(select, (db_key, raw, time.time())).fetchall()

if not rows:

if self.statistics:

sql(cache_miss)

return default

(rowid, db_expire_time, db_tag,

mode, filename, db_value), = rows

try:

value = self._disk.fetch(mode, filename, db_value, read)

except IOError as error:

if error.errno == errno.ENOENT:

# Key was deleted before we could retrieve result.

if self.statistics:

sql(cache_miss)

return default

else:

raise

if self.statistics:

sql(cache_hit)

if update_column is not None:

sql(update % update_column, (rowid,))

if expire_time and tag:

return (value, db_expire_time, db_tag)

elif expire_time:

return (value, db_expire_time)

elif tag:

return (value, db_tag)

else:

return value

def __getitem__(self, key):

"""Return corresponding value for `key` from cache.

:param key: key matching item

:return: corresponding value

:raises KeyError: if key is not found

:raises Timeout: if database timeout expires

"""

value = self.get(key, default=ENOVAL)

if value is ENOVAL:

raise KeyError(key)

return value

def read(self, key):

"""Return file handle value corresponding to `key` from cache.

:param key: key matching item

:return: file open for reading in binary mode

:raises KeyError: if key is not found

:raises Timeout: if database timeout expires

"""

handle = self.get(key, default=ENOVAL, read=True)

if handle is ENOVAL:

raise KeyError(key)

return handle

def __contains__(self, key):

"""Return `True` if `key` matching item is found in cache.

:param key: key matching item

:return: True if key matching item

"""

sql = self._sql

db_key, raw = self._disk.put(key)