"""Core disk and file backed cache API.

"""

import codecs

import contextlib as cl

import errno

import functools as ft

import io

import json

import os

import os.path as op

import pickle

import pickletools

import sqlite3

import struct

import tempfile

import threading

import time

import warnings

import zlib

def full_name(func):

"""Return full name of `func` by adding the module and function name."""

return func.__module__ + '.' + func.__qualname__

class Constant(tuple):

"""Pretty display of immutable constant."""

def __new__(cls, name):

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

def __repr__(self):

return '%s' % self[0]

DBNAME = 'cache.db'

ENOVAL = Constant('ENOVAL')

UNKNOWN = Constant('UNKNOWN')

MODE_NONE = 0

MODE_RAW = 1

MODE_BINARY = 2

MODE_TEXT = 3

MODE_PICKLE = 4

DEFAULT_SETTINGS = {

'statistics': 0, # False

'tag_index': 0, # False

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

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

'cull_limit': 10,

'sqlite_auto_vacuum': 1, # FULL

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

'sqlite_journal_mode': 'wal',

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

'sqlite_synchronous': 1, # NORMAL

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

'disk_pickle_protocol': pickle.HIGHEST_PROTOCOL,

}

METADATA = {

'count': 0,

'size': 0,

'hits': 0,

'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 {fields} 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 = {now}',

'cull': 'SELECT {fields} 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 {fields} FROM Cache ORDER BY access_count LIMIT ?',

},

}

class Disk:

"""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 str:

return zlib.adler32(disk_key.encode('utf-8')) & mask # noqa

elif type_disk_key is int:

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=unidiomatic-typecheck

type_key = type(key)

if type_key is bytes:

return sqlite3.Binary(key), True

elif (

(type_key is str)

or (

type_key is int

and -9223372036854775808 <= key <= 9223372036854775807

)

or (type_key is float)

):

return key, True

else:

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

result = pickletools.optimize(data)

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=unidiomatic-typecheck

if raw:

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

else:

return pickle.load(io.BytesIO(key))

def store(self, value, read, key=UNKNOWN):

"""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

:param key: key for item (default UNKNOWN)

: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 str and len(value) < min_file_size)

or (

type_value is int

and -9223372036854775808 <= value <= 9223372036854775807

)

or (type_value is float)

):

return 0, MODE_RAW, None, value

elif type_value is bytes:

if len(value) < min_file_size:

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

else:

filename, full_path = self.filename(key, value)

self._write(full_path, io.BytesIO(value), 'xb')

return len(value), MODE_BINARY, filename, None

elif type_value is str:

filename, full_path = self.filename(key, value)

self._write(full_path, io.StringIO(value), 'x', 'UTF-8')

size = op.getsize(full_path)

return size, MODE_TEXT, filename, None

elif read:

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

filename, full_path = self.filename(key, value)

iterator = iter(reader, b'')

size = self._write(full_path, iterator, 'xb')

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(key, value)

self._write(full_path, io.BytesIO(result), 'xb')

return len(result), MODE_PICKLE, filename, None

def _write(self, full_path, iterator, mode, encoding=None):

full_dir, _ = op.split(full_path)

for count in range(1, 11):

with cl.suppress(OSError):

os.makedirs(full_dir)

try:

# Another cache may have deleted the directory before

# the file could be opened.

writer = open(full_path, mode, encoding=encoding)

except OSError:

if count == 10:

# Give up after 10 tries to open the file.

raise

continue

with writer:

size = 0

for chunk in iterator:

size += len(chunk)

writer.write(chunk)

return size

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

:raises: IOError if the value cannot be read

"""

# pylint: disable=unidiomatic-typecheck,consider-using-with

if mode == MODE_RAW:

return bytes(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 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(io.BytesIO(value))

def filename(self, key=UNKNOWN, value=UNKNOWN):

"""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 may be slow.

The default implementation ignores the `key` and `value` parameters.

In some scenarios, for example :meth:`Cache.push

<diskcache.Cache.push>`, the `key` or `value` may not be known when the

item is stored in the cache.

:param key: key for item (default UNKNOWN)

:param value: value for item (default UNKNOWN)

"""

# pylint: disable=unused-argument

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'

filename = op.join(sub_dir, name)

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

return filename, full_path

def remove(self, file_path):

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

This method is cross-thread and cross-process safe. If an OSError

occurs, it is suppressed.

:param str file_path: relative path to file

"""

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

full_dir, _ = op.split(full_path)

# Suppress OSError that may occur if two caches attempt to delete the

# same file or directory at the same time.

with cl.suppress(OSError):

os.remove(full_path)

with cl.suppress(OSError):

os.removedirs(full_dir)

class JSONDisk(Disk):

"""Cache key and value using JSON serialization with zlib compression."""

def __init__(self, directory, compress_level=1, **kwargs):

"""Initialize JSON disk instance.

Keys and values are compressed using the zlib library. The

`compress_level` is an integer from 0 to 9 controlling the level of

compression; 1 is fastest and produces the least compression, 9 is

slowest and produces the most compression, and 0 is no compression.

:param str directory: directory path

:param int compress_level: zlib compression level (default 1)

:param kwargs: super class arguments

"""

self.compress_level = compress_level

super().__init__(directory, **kwargs)

def put(self, key):

json_bytes = json.dumps(key).encode('utf-8')

data = zlib.compress(json_bytes, self.compress_level)

return super().put(data)

def get(self, key, raw):

data = super().get(key, raw)

return json.loads(zlib.decompress(data).decode('utf-8'))

def store(self, value, read, key=UNKNOWN):

if not read:

json_bytes = json.dumps(value).encode('utf-8')

value = zlib.compress(json_bytes, self.compress_level)

return super().store(value, read, key=key)

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

data = super().fetch(mode, filename, value, read)

if not read:

data = json.loads(zlib.decompress(data).decode('utf-8'))

return data

class Timeout(Exception):

"""Database timeout expired."""

class UnknownFileWarning(UserWarning):

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

class EmptyDirWarning(UserWarning):

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

def args_to_key(base, args, kwargs, typed, ignore):

"""Create cache key out of function arguments.

:param tuple base: base of key

:param tuple args: function arguments

:param dict kwargs: function keyword arguments

:param bool typed: include types in cache key

:param set ignore: positional or keyword args to ignore

:return: cache key tuple

"""

args = tuple(arg for index, arg in enumerate(args) if index not in ignore)

key = base + args + (None,)

if kwargs:

kwargs = {key: val for key, val in kwargs.items() if key not in ignore}

sorted_items = sorted(kwargs.items())

for item in sorted_items:

key += item

if typed:

key += tuple(type(arg) for arg in args)

if kwargs:

key += tuple(type(value) for _, value in sorted_items)

return key

class Cache:

"""Disk and file backed cache."""

def __init__(self, directory=None, 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') from None

if directory is None:

directory = tempfile.mkdtemp(prefix='diskcache-')

directory = str(directory)

directory = op.expanduser(directory)

directory = op.expandvars(directory)

self._directory = directory

self._timeout = 0 # Manually handle retries during initialization.

self._local = threading.local()

self._txn_id = None

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,

) from None

sql = self._sql_retry

# Setup Settings table.

try:

current_settings = dict(

sql('SELECT key, value FROM Settings').fetchall()

)

except sqlite3.OperationalError:

current_settings = {}

sets = DEFAULT_SETTINGS.copy()

sets.update(current_settings)

sets.update(settings)

for key in METADATA:

sets.pop(key, None)

# Chance to set pragmas before any tables are created.

for key, value in sorted(sets.items()):

if key.startswith('sqlite_'):

self.reset(key, value, update=False)

sql(

'CREATE TABLE IF NOT EXISTS Settings ('

' key TEXT NOT NULL UNIQUE,'

' value)'

)

# 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,'

' 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

self._sql # pylint: disable=pointless-statement

@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 _con(self):

# Check process ID to support process forking. If the process

# ID changes, close the connection and update the process ID.

local_pid = getattr(self._local, 'pid', None)

pid = os.getpid()

if local_pid != pid:

self.close()

self._local.pid = pid

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,

)

# Some SQLite pragmas work on a per-connection basis so

# query the Settings table and reset the pragmas. The

# Settings table may not exist so catch and ignore the

# OperationalError that may occur.

try:

select = 'SELECT key, value FROM Settings'

settings = con.execute(select).fetchall()

except sqlite3.OperationalError:

pass

else:

for key, value in settings:

if key.startswith('sqlite_'):

self.reset(key, value, update=False)

return con

@property

def _sql(self):

return self._con.execute

@property

def _sql_retry(self):

sql = self._sql

# 2018-11-01 GrantJ - Some SQLite builds/versions handle

# the SQLITE_BUSY return value and connection parameter

# "timeout" differently. For a more reliable duration,

# manually retry the statement for 60 seconds. Only used

# by statements which modify the database and do not use

# a transaction (like those in ``__init__`` or ``reset``).

# See Issue #85 for and tests/issue_85.py for more details.

def _execute_with_retry(statement, *args, **kwargs):

start = time.time()

while True:

try:

return sql(statement, *args, **kwargs)

except sqlite3.OperationalError as exc:

if str(exc) != 'database is locked':

raise

diff = time.time() - start

if diff > 60:

raise

time.sleep(0.001)

return _execute_with_retry

@cl.contextmanager

def transact(self, retry=False):

"""Context manager to perform a transaction by locking the cache.

While the cache is locked, no other write operation is permitted.

Transactions should therefore be as short as possible. Read and write

operations performed in a transaction are atomic. Read operations may

occur concurrent to a transaction.

Transactions may be nested and may not be shared between threads.

Raises :exc:`Timeout` error when database timeout occurs and `retry` is

`False` (default).

>>> cache = Cache()

>>> with cache.transact(): # Atomically increment two keys.

... _ = cache.incr('total', 123.4)

... _ = cache.incr('count', 1)

>>> with cache.transact(): # Atomically calculate average.

... average = cache['total'] / cache['count']

>>> average

123.4

:param bool retry: retry if database timeout occurs (default False)

:return: context manager for use in `with` statement

:raises Timeout: if database timeout occurs

"""

with self._transact(retry=retry):

yield

@cl.contextmanager

def _transact(self, retry=False, filename=None):

sql = self._sql

filenames = []

_disk_remove = self._disk.remove

tid = threading.get_ident()

txn_id = self._txn_id

if tid == txn_id:

begin = False

else:

while True:

try:

sql('BEGIN IMMEDIATE')

begin = True

self._txn_id = tid

break

except sqlite3.OperationalError:

if retry:

continue

if filename is not None:

_disk_remove(filename)

raise Timeout from None

try:

yield sql, filenames.append

except BaseException:

if begin:

assert self._txn_id == tid

self._txn_id = None

sql('ROLLBACK')

raise

else:

if begin:

assert self._txn_id == tid

self._txn_id = None

sql('COMMIT')

for name in filenames:

if name is not None:

_disk_remove(name)

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

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

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

for reading in binary mode.

Raises :exc:`Timeout` error when database timeout occurs and `retry` is

`False` (default).

: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)

:param bool retry: retry if database timeout occurs (default False)

:return: True if item was set

:raises Timeout: if database timeout occurs

"""

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, key=key)

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(retry, 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

def __setitem__(self, key, value):

"""Set corresponding `value` for `key` in cache.

:param key: key for item

:param value: value for item

:return: corresponding value

:raises KeyError: if key is not found

"""

self.set(key, value, retry=True)

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

sql = self._sql

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

sql(

'UPDATE Cache SET'

' store_time = ?,'

' expire_time = ?,'

' access_time = ?,'

' access_count = ?,'

' tag = ?,'

' size = ?,'

' mode = ?,'

' filename = ?,'

' value = ?'

' WHERE rowid = ?',

(

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, store_time, expire_time, access_time,'

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

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

(

key,

raw,

now, # store_time

expire_time,

now, # access_time

0, # access_count

tag,

size,

mode,

filename,

value,

),

)

def _cull(self, now, sql, cleanup, limit=None):

cull_limit = self.cull_limit if limit is None else 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 = EVICTION_POLICY[self.eviction_policy]['cull']

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

return

select_filename = select_policy.format(fields='filename', now=now)

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

if rows:

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

select_policy.format(fields='rowid', now=now)

)

sql(delete, (cull_limit,))

for (filename,) in rows:

cleanup(filename)

def touch(self, key, expire=None, retry=False):

"""Touch `key` in cache and update `expire` time.

Raises :exc:`Timeout` error when database timeout occurs and `retry` is

`False` (default).

:param key: key for item

:param float expire: seconds until item expires

(default None, no expiry)

:param bool retry: retry if database timeout occurs (default False)

:return: True if key was touched

:raises Timeout: if database timeout occurs

"""

now = time.time()

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

expire_time = None if expire is None else now + expire

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

rows = sql(

'SELECT rowid, expire_time FROM Cache'

' WHERE key = ? AND raw = ?',

(db_key, raw),

).fetchall()

if rows:

((rowid, old_expire_time),) = rows

if old_expire_time is None or old_expire_time > now:

sql(

'UPDATE Cache SET expire_time = ? WHERE rowid = ?',

(expire_time, rowid),

)

return True

return False

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

"""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.

Raises :exc:`Timeout` error when database timeout occurs and `retry` is

`False` (default).

: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)

:param bool retry: retry if database timeout occurs (default False)

:return: True if item was added

:raises Timeout: if database timeout occurs

"""

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, key=key)

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

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

rows = sql(

'SELECT rowid, filename, expire_time FROM Cache'

' WHERE key = ? AND raw = ?',

(db_key, raw),

).fetchall()