Source code for hypothesis.extra.redis

# This file is part of Hypothesis, which may be found at
# Most of this work is copyright (C) 2013-2020 David R. MacIver
# (, but it contains contributions by others. See
# CONTRIBUTING.rst for a full list of people who may hold copyright, and
# consult the git log if you need to determine who owns an individual
# contribution.
# This Source Code Form is subject to the terms of the Mozilla Public License,
# v. 2.0. If a copy of the MPL was not distributed with this file, You can
# obtain one at

from contextlib import contextmanager
from datetime import timedelta
from typing import Iterable

from redis import Redis

from hypothesis.database import ExampleDatabase
from hypothesis.internal.validation import check_type

[docs]class RedisExampleDatabase(ExampleDatabase): """Store Hypothesis examples as sets in the given :class:`redis.Redis` datastore. This is particularly useful for shared databases, as per the recipe for a :class:`~hypothesis.database.MultiplexedDatabase`. .. note:: If a test has not been run for ``expire_after``, those examples will be allowed to expire. The default time-to-live persists examples between weekly runs. """ def __init__( self, redis: Redis, *, expire_after: timedelta = timedelta(days=8), key_prefix: bytes = b"hypothesis-example:", ): check_type(Redis, redis, "redis") check_type(timedelta, expire_after, "expire_after") check_type(bytes, key_prefix, "key_prefix") self.redis = redis self._expire_after = expire_after self._prefix = key_prefix def __repr__(self) -> str: return ( f"RedisExampleDatabase({self.redis!r}, expire_after={self._expire_after!r})" ) @contextmanager def _pipeline(self, *reset_expire_keys, transaction=False, auto_execute=True): # Context manager to batch updates and expiry reset, reducing TCP roundtrips pipe = self.redis.pipeline(transaction=transaction) yield pipe for key in reset_expire_keys: pipe.expire(self._prefix + key, self._expire_after) if auto_execute: pipe.execute() def fetch(self, key: bytes) -> Iterable[bytes]: with self._pipeline(key, auto_execute=False) as pipe: pipe.smembers(self._prefix + key) yield from pipe.execute()[0] def save(self, key: bytes, value: bytes) -> None: with self._pipeline(key) as pipe: pipe.sadd(self._prefix + key, value) def delete(self, key: bytes, value: bytes) -> None: with self._pipeline(key) as pipe: pipe.srem(self._prefix + key, value) def move(self, src: bytes, dest: bytes, value: bytes) -> None: with self._pipeline(src, dest) as pipe: pipe.srem(self._prefix + src, value) pipe.sadd(self._prefix + dest, value)