python-redis-orm

python-redis-orm

Python Redis ORM library that gives redis easy-to-use objects with fields and speeds a development up, inspired by Django ORM

For one project, I needed to work with redis, but redis-py provides a minimum level of work with redis. I didn't find any Django-like ORM for redis, so I wrote this library, then there will be a port to Django.

Working with this library, you are expected:

• Fully works in 2021
• Django-like architecture
• Easy adaptation to your needs
• Adequate informational messages and error messages
• Built-in RedisRoot class that stores specified models, with:
  • redis_instance setting - your redis connection (from redis-py)
  • prefix setting - prefix of this RedisRoot to be stored in redis
  • ignore_deserialization_errors setting - do not raise errors, while deserializing data
  • save_consistency setting - show structure-first data
  • economy setting - to not return full data and save some requests (usually, speeds up your app on 80%)
• 13 built-in types of fields:
  • RedisField - base class for nesting
  • RedisString - string
  • RedisNumber - int or float
  • RedisId - instances IDs
  • RedisBool - bool
  • RedisDecimal - working accurately with numbers via decimal
  • RedisJson - for data, that can be JSONed
  • RedisList - list
  • RedisDict - dict
  • RedisDateTime - for work with date and time, via python datetime.datetime
  • RedisDate - for work with date, via python datetime.data
  • RedisForeignKey - for link to other instance
  • RedisManyToMany - for links to other instances
• All fields supports:
  • Automatically serialization
  • Automatically deserialization
  • TTL (Time To Live) setting
  • Default values
  • Providing functions without call, to call, while need
  • Allow null values setting
  • Choices
  • Filtering (and deep filtering):
    • exact - equality
    • iexact - case-independent equality
    • contains - is filter string in the value string
    • icontains - is filter string case-independent in the value string
    • in - is value in the provided list
    • gt - is value greater
    • gte - is value greater or equals
    • lt - is value less
    • lte - is value less or equals
    • startswith - is string starts with
    • istartswith - is string case-independent starts with
    • endswith - is string ends with
    • iendswith - is string case-independent ends wth
    • range - is value in provided range
    • isnull - is value in ["null", None]
• Built-in RedisModel class, with:
  • All fields that you want
  • TTL (Time To Live)
• CRUD (Create Read Update Delete)
• Non-blocking usage! Any operation gives the same result as the default, but it just creates an asyncio task in the background instead of write inside the call

Installation

pip install python-redis-orm

Here is PyPI

Obviously, you need to install and run redis server on your machine, we support v3+

Usage

1. Create RedisRoot with params:
   • prefix (str) - prefix for your redis root
   • connection_pool (redis.ConnectionPool) - redis-py redis.ConnectionPool instance, with decode_responses=True
   • ignore_deserialization_errors (bool) - to ignore deserialization errors or raise exception
   • save_consistency (bool) - to use structure-first data
   • economy (bool) - if True, all update requests will return only instance id
   • use_keys (bool) - to use Redis keys command (uses memory instead of CPU) instead of scan
2. Create your models
3. Call register_models() on your RedisRoot instance and provide list with your models
4. Use our CRUD

CRUD

example_instance = ExampleModel(example_field='example_data').save() # - to create an instance and get its data dict
# or:
example_instance = redis_root.create(ExampleModel, example_field='example_data')
filtered_example_instances = redis_root.get(ExampleModel, example_field='example_data') # - to get all ExampleModel instances with example_field filter and get its data dict
ordered_instances = redis_root.order(filtered_example_instances, '-id') # - to get ordered filtered_example_instances by id ('-' for reverse)
updated_example_instances = redis_root.update(ExampleModel, ordered_instances, example_field='another_example_data') # - to update all ordered_instances example_field with value 'another_example_data' and get its data dict
redis_root.delete(ExampleModel, updated_example_instances) # - to delete updated_example_instances

# Non-blocking funcs are the same, just add "_nb" to the end:
# ExampleModel(...).save_nb()
# redis_root.create_nb(...)
# redis_root.update_nb(...)
# redis_root.delete_nb(...)

Example usage

All features:

full_test.py

import random
import sys
from time import sleep
import asyncio
import os

from python_redis_orm.core import *


def generate_token(chars_count):
    allowed_chars = 'QWERTYUIOPASDFGHJKLZXCVBNM1234567890'
    token = f'{"".join([random.choice(allowed_chars) for i in range(chars_count)])}'
    return token


def generate_token_12_chars():
    return generate_token(12)


class BotSession(RedisModel):
    session_token = RedisString(default=generate_token_12_chars)
    created = RedisDateTime(default=datetime.datetime.now)


class TaskChallenge(RedisModel):
    bot_session = RedisForeignKey(model=BotSession)
    task_id = RedisNumber(default=0, null=False)
    status = RedisString(default='in_work', choices={
        'in_work': 'В работе',
        'completed': 'Завершён успешно',
        'completed_frozen_points': 'Завершён успешно, получил поинты в холде',
        'completed_points': 'Завершён успешно, получил поинты',
        'completed_decommissioning': 'Завершён успешно, поинты списаны',
        'failed_bot': 'Зафейлил бот',
        'failed_task_creator': 'Зафейлил создатель задания',
    }, null=False)
    account_checks_count = RedisNumber(default=0)
    created = RedisDateTime(default=datetime.datetime.now)


class TtlCheckModel(RedisModel):
    redis_number_with_ttl = RedisNumber(default=0, null=False)


class MetaTtlCheckModel(RedisModel):
    redis_number = RedisNumber(default=0, null=False)
    
    class Meta:
        ttl = 5


class DictCheckModel(RedisModel):
    redis_dict = RedisDict()


class ListCheckModel(RedisModel):
    redis_list = RedisList()


class ForeignKeyCheckModel(RedisModel):
    task_challenge = RedisForeignKey(model=TaskChallenge)


class ManyToManyCheckModel(RedisModel):
    task_challenges = RedisManyToMany(model=TaskChallenge)


class ModelWithOverriddenSave(RedisModel):
    multiplied_max_field = RedisNumber()
    
    def save(self):
        redis_root = self.get('redis_root')  # get value of any field
        new_value = 1
        all_instances = redis_root.get(ModelWithOverriddenSave)
        if all_instances:
            max_value = max(list(map(lambda instance: instance['multiplied_max_field'], all_instances)))
            new_value = max_value * 2
        self.set(multiplied_max_field=new_value)
        return super().save()


def clean_db_after_test(connection_pool, prefix):
    redis_instance = redis.Redis(connection_pool=connection_pool)
    for key in redis_instance.keys(f'{prefix}:*'):
        redis_instance.delete(key)


def basic_test(connection_pool, prefix):
    try:
        redis_root = RedisRoot(
            prefix=prefix,
            connection_pool=connection_pool,
            ignore_deserialization_errors=True
        )
        redis_root.register_models([
            TaskChallenge,
        ])
        for i in range(5):
            TaskChallenge(
                redis_root=redis_root,
                status='in_work',
            ).save()
        task_challenges_without_keys = redis_root.get(TaskChallenge)
        task_challenges_with_keys = redis_root.get(TaskChallenge, return_dict=True)
        have_exception = False
        if not len(task_challenges_without_keys):
            have_exception = True
        if not task_challenges_with_keys:
            have_exception = True
        else:
            if not task_challenges_with_keys.keys():
                have_exception = True
            else:
                if len(list(task_challenges_with_keys.keys())) != len(task_challenges_without_keys):
                    have_exception = True
    except BaseException as ex:
        have_exception = True
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def auto_reg_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True
    )
    task_challenge_1 = TaskChallenge(
        redis_root=redis_root,
        status='in_work',
    ).save()
    try:
        task_challenges = redis_root.get(TaskChallenge)
        have_exception = False
    except BaseException as ex:
        have_exception = True
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def no_connection_pool_test(*args, **kwargs):
    try:
        redis_root = RedisRoot(
            ignore_deserialization_errors=True
        )
        task_challenge_1 = TaskChallenge(
            redis_root=redis_root,
            status='in_work',
        )
        task_challenge_1.save()
        task_challenges = redis_root.get(TaskChallenge)
        have_exception = False
        connection_pool = redis.ConnectionPool(
            host=os.environ['REDIS_HOST'],
            port=os.environ['REDIS_PORT'],
            db=0,
            decode_responses=True
        )
        clean_db_after_test(connection_pool, redis_root.prefix)
    except BaseException as ex:
        have_exception = True
    return have_exception


def choices_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True
    )
    task_challenge_1 = TaskChallenge(
        redis_root=redis_root,
        status='bruh',
    )
    try:
        save_result = task_challenge_1.save()
        task_challenges = redis_root.get(TaskChallenge)
        have_exception = True
    except BaseException as ex:
        have_exception = False
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def order_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True
    )
    for i in range(3):
        TaskChallenge(
            redis_root=redis_root
        ).save()
    have_exception = True
    try:
        task_challenges = redis_root.get(TaskChallenge)
        first_task_challenge = redis_root.order(task_challenges, 'id')[0]
        last_task_challenge = redis_root.order(task_challenges, '-id')[0]
        if first_task_challenge['id'] == 1 and last_task_challenge['id'] == len(task_challenges):
            have_exception = False
    except BaseException as ex:
        pass
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def filter_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True
    )
    have_exception = True
    try:
        same_tokens_count = 2
        random_tokens_count = 8
        same_token = generate_token(50)
        random_tokens = [generate_token(50) for i in range(random_tokens_count)]
        for i in range(same_tokens_count):
            BotSession(redis_root, session_token=same_token).save()
        for random_token in random_tokens:
            BotSession(redis_root, session_token=random_token).save()
        task_challenges_with_same_token = redis_root.get(BotSession, session_token=same_token)
        if len(task_challenges_with_same_token) == same_tokens_count:
            have_exception = False
    except BaseException as ex:
        print(ex)
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def update_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True
    )
    have_exception = True
    try:
        bot_session_1 = BotSession(redis_root, session_token='123').save()
        bot_session_1_id = bot_session_1['id']
        redis_root.update(BotSession, bot_session_1, session_token='234')
        bot_sessions_filtered = redis_root.get(BotSession, id=bot_session_1_id)
        if len(bot_sessions_filtered) == 1:
            bot_session_1_new = bot_sessions_filtered[0]
            if 'session_token' in bot_session_1_new.keys():
                if bot_session_1_new['session_token'] == '234':
                    have_exception = False
    except BaseException as ex:
        print(ex)
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def functions_like_defaults_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True
    )
    have_exception = False
    try:
        bot_session_1 = BotSession(redis_root).save()
        bot_session_2 = BotSession(redis_root).save()
        if bot_session_1.session_token == bot_session_2.session_token:
            have_exception = True
    except BaseException as ex:
        pass
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def redis_foreign_key_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True
    )
    have_exception = True
    try:
        bot_session_1 = BotSession(
            redis_root=redis_root,
        ).save()
        task_challenge_1 = TaskChallenge(
            redis_root=redis_root,
            bot_session=bot_session_1
        ).save()
        bot_sessions = redis_root.get(BotSession)
        bot_session = redis_root.order(bot_sessions, '-id')[0]
        task_challenges = redis_root.get(TaskChallenge)
        task_challenge = redis_root.order(task_challenges, '-id')[0]
        if type(task_challenge['bot_session']) == dict:
            if task_challenge['bot_session'] == bot_session:
                have_exception = False
    except BaseException as ex:
        print(ex)
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def delete_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True
    )
    have_exception = True
    try:
        bot_session_1 = BotSession(
            redis_root=redis_root,
        ).save()
        task_challenge_1 = TaskChallenge(
            redis_root=redis_root,
            bot_session=bot_session_1
        ).save()
        redis_root.delete(BotSession, bot_session_1)
        redis_root.delete(TaskChallenge, task_challenge_1)
        bot_sessions = redis_root.get(BotSession)
        task_challenges = redis_root.get(TaskChallenge)
        if len(bot_sessions) == 0 and len(task_challenges) == 0:
            have_exception = False
    except BaseException as ex:
        print(ex)
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def save_consistency_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True,
        save_consistency=True,
    )
    have_exception = True
    try:
        ttl_check_model_1 = TtlCheckModel(
            redis_root=redis_root,
        ).save()
        ttl_check_models = redis_root.get(TtlCheckModel)
        if len(ttl_check_models):
            ttl_check_model = ttl_check_models[0]
            if 'redis_number_with_ttl' in ttl_check_model.keys():
                sleep(6)
                ttl_check_models = redis_root.get(TtlCheckModel)
                if len(ttl_check_models):
                    ttl_check_model = ttl_check_models[0]
                    if 'redis_number_with_ttl' in ttl_check_model.keys():  # because consistency is saved
                        have_exception = False
    except BaseException as ex:
        print(ex)
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def meta_ttl_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True,
    )
    have_exception = True
    try:
        meta_ttl_check_model_1 = MetaTtlCheckModel(
            redis_root=redis_root,
        ).save()
        meta_ttl_check_models = redis_root.get(MetaTtlCheckModel)
        if len(meta_ttl_check_models):
            meta_ttl_check_model = meta_ttl_check_models[0]
            if 'redis_number' in meta_ttl_check_model.keys():
                sleep(6)
                meta_ttl_check_models = redis_root.get(MetaTtlCheckModel)
                if not len(meta_ttl_check_models):
                    have_exception = False
    except BaseException as ex:
        print(ex)
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def use_keys_test(connection_pool, prefix):
    have_exception = True
    try:
        
        redis_root = RedisRoot(
            prefix=prefix,
            connection_pool=connection_pool,
            ignore_deserialization_errors=True,
            use_keys=True
        )
        started_in_keys = datetime.datetime.now()
        tests_count = 100
        for i in range(tests_count):
            task_challenge_1 = TaskChallenge(
                redis_root=redis_root,
                status='in_work',
            ).save()
            redis_root.update(TaskChallenge, task_challenge_1, account_checks_count=1)
        ended_in_keys = datetime.datetime.now()
        keys_time = (ended_in_keys - started_in_keys).total_seconds()
        clean_db_after_test(connection_pool, prefix)
        
        redis_root = RedisRoot(
            prefix=prefix,
            connection_pool=connection_pool,
            ignore_deserialization_errors=True,
            use_keys=False
        )
        started_in_no_keys = datetime.datetime.now()
        for i in range(tests_count):
            task_challenge_1 = TaskChallenge(
                redis_root=redis_root,
                status='in_work',
            ).save()
            redis_root.update(TaskChallenge, task_challenge_1, account_checks_count=1)
        ended_in_no_keys = datetime.datetime.now()
        no_keys_time = (ended_in_no_keys - started_in_no_keys).total_seconds()
        clean_db_after_test(connection_pool, prefix)
        keys_percent = round((no_keys_time / keys_time - 1) * 100, 2)
        keys_symbol = ('+' if keys_percent > 0 else '')
        print(f'Keys usage gives {keys_symbol}{keys_percent}% efficiency')
        have_exception = False
    except BaseException as ex:
        print(ex)
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def dict_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True
    )
    have_exception = True
    try:
        some_dict = {
            'age': 19,
            'weed': True
        }
        DictCheckModel(
            redis_root=redis_root,
            redis_dict=some_dict
        ).save()
        dict_check_model_instance = redis_root.get(DictCheckModel)[0]
        if 'redis_dict' in dict_check_model_instance.keys():
            if dict_check_model_instance['redis_dict'] == some_dict:
                have_exception = False
    except BaseException as ex:
        print(ex)
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def list_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True
    )
    have_exception = True
    try:
        some_list = [5, 9, 's', 4.5, False]
        ListCheckModel(
            redis_root=redis_root,
            redis_list=some_list
        ).save()
        list_check_model_instance = redis_root.get(ListCheckModel)[0]
        if 'redis_list' in list_check_model_instance.keys():
            if list_check_model_instance['redis_list'] == some_list:
                have_exception = False
    except BaseException as ex:
        print(ex)
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def non_blocking_test(connection_pool, prefix):
    have_exception = True
    
    # try:
    
    def task(data_count, use_non_blocking):
        connection_pool = redis.ConnectionPool(
            host=os.environ['REDIS_HOST'],
            port=os.environ['REDIS_PORT'],
            db=0,
            decode_responses=True
        )
        redis_root = RedisRoot(
            prefix=prefix,
            connection_pool=connection_pool,
            ignore_deserialization_errors=True
        )
        
        for i in range(data_count):
            redis_root.create(
                ListCheckModel,
                redis_list=['update_list']
            )
            redis_root.create(
                ListCheckModel,
                redis_list=['delete_list']
            )
        
        def create_list():
            if use_non_blocking:
                list_check_model_instance = redis_root.create_nb(
                    ListCheckModel,
                    redis_list=['create_list']
                )
            else:
                list_check_model_instance = redis_root.create(
                    ListCheckModel,
                    redis_list=['create_list']
                )
        
        def update_list():
            to_update = redis_root.get(
                ListCheckModel,
                redis_list=['update_list']
            )
            if use_non_blocking:
                updated_instance = redis_root.update_nb(
                    ListCheckModel,
                    to_update,
                    redis_list=['now_updated_list']
                )
            else:
                updated_instance = redis_root.update(
                    ListCheckModel,
                    to_update,
                    redis_list=['now_updated_list']
                )
        
        def delete_list():
            to_delete = redis_root.get(
                ListCheckModel,
                redis_list=['delete_list']
            )
            if use_non_blocking:
                redis_root.delete_nb(
                    ListCheckModel,
                    to_delete,
                )
            else:
                redis_root.delete(
                    ListCheckModel,
                    to_delete,
                )
        
        tests = [
            create_list,
            update_list,
            delete_list,
        ]
        for test in tests:
            for i in range(data_count):
                test()
    
    data_count = 100
    clean_db_after_test(connection_pool, prefix)
    nb_started_in = datetime.datetime.now()
    task(data_count, True)
    nb_ended_in = datetime.datetime.now()
    nb_time = (nb_ended_in - nb_started_in).total_seconds()
    clean_db_after_test(connection_pool, prefix)
    b_started_in = datetime.datetime.now()
    task(data_count, False)
    b_ended_in = datetime.datetime.now()
    b_time = (b_ended_in - b_started_in).total_seconds()
    clean_db_after_test(connection_pool, prefix)
    
    nb_percent = round((nb_time / b_time - 1) * 100, 2)
    nb_symbol = ('+' if nb_percent > 0 else '')
    print(f'Non blocking gives {nb_symbol}{nb_percent}% efficiency')
    have_exception = False
    # except BaseException as ex:
    #     print(ex)
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def foreign_key_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True,
    )
    have_exception = False
    try:
        task_id = 12345
        task_challenge = TaskChallenge(
            redis_root=redis_root,
            task_id=task_id
        ).save()
        foreign_key_check_instance = redis_root.create(
            ForeignKeyCheckModel,
            task_challenge=task_challenge
        )
        # Check really created
        task_challenge_qs = redis_root.get(TaskChallenge, task_id=task_id)
        if len(task_challenge_qs) != 1:
            have_exception = True
        else:
            task_challenge = task_challenge_qs[0]
            foreign_key_check_instance_qs = redis_root.get(ForeignKeyCheckModel, task_challenge=task_challenge)
            if len(foreign_key_check_instance_qs) != 1:
                have_exception = True
            else:
                foreign_key_check_instance = foreign_key_check_instance_qs[0]
                if foreign_key_check_instance['task_challenge']['task_id'] != task_id:
                    have_exception = True
    except BaseException as ex:
        print(ex)
        have_exception = True
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def many_to_many_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True,
    )
    have_exception = False
    try:
        tasks_ids = set([random.randrange(0, 100) for i in range(10)])
        task_challenges = [
            TaskChallenge(
                redis_root=redis_root,
                task_id=task_id
            ).save()
            for task_id in tasks_ids
        ]
        many_to_many_check_instance = redis_root.create(
            ManyToManyCheckModel,
            task_challenges=task_challenges
        )
        # Check really created
        many_to_many_check_instances_qs = redis_root.get(ManyToManyCheckModel)
        if len(many_to_many_check_instances_qs) != 1:
            have_exception = True
        else:
            many_to_many_check_instance = many_to_many_check_instances_qs[0]
            if many_to_many_check_instance['task_challenges'] != task_challenges:
                have_exception = True
    except BaseException as ex:
        print(ex)
        have_exception = True
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def save_override_test(connection_pool, prefix):
    redis_root = RedisRoot(
        prefix=prefix,
        connection_pool=connection_pool,
        ignore_deserialization_errors=True,
    )
    have_exception = False
    try:
        instance_1 = redis_root.create(ModelWithOverriddenSave)
        instance_2 = redis_root.create(ModelWithOverriddenSave)
        if instance_1['multiplied_max_field'] * 2 != instance_2['multiplied_max_field']:
            have_exception = True
    except BaseException as ex:
        print(ex)
        have_exception = True
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def performance_test(connection_pool, prefix):
    have_exception = False
    try:
        
        def run_test(count, model):
            
            def test(count, model, **test_params):
                real_test_params = {
                    'use_keys': True,
                    'use_non_blocking': True
                }
                for key in real_test_params.copy():
                    if key in test_params.keys():
                        real_test_params[key] = test_params[key]
                
                def create_instances(redis_root, count, use_non_blocking, model):
                    
                    if use_non_blocking:
                        started_in = datetime.datetime.now()
                        results = [
                            redis_root.create_nb(model)
                            for i in range(count)
                        ]
                        ended_in = datetime.datetime.now()
                    else:
                        started_in = datetime.datetime.now()
                        results = [
                            redis_root.create(model)
                            for i in range(count)
                        ]
                        ended_in = datetime.datetime.now()
                    
                    time_took = (ended_in - started_in).total_seconds()
                    fields_count = len(results[0].keys()) * count
                    clean_db_after_test(connection_pool, prefix)
                    return [time_took, count, fields_count]
                
                redis_root = RedisRoot(
                    prefix=prefix,
                    connection_pool=connection_pool,
                    ignore_deserialization_errors=True,
                    use_keys=real_test_params['use_keys']
                )
                
                test_result = create_instances(redis_root, count, real_test_params['use_non_blocking'], model)
                
                return test_result
            
            test_confs = [
                {
                    'use_keys': False,
                    'use_non_blocking': False,
                },
                {
                    'use_keys': False,
                    'use_non_blocking': True,
                },
                {
                    'use_keys': True,
                    'use_non_blocking': False,
                },
                {
                    'use_keys': True,
                    'use_non_blocking': True,
                },
            
            ]
            
            test_confs_results = [
                test(count, model, **test_conf)
                for test_conf in test_confs
            ]
            
            print(f'\n\n\n'
                  f'Performance test results on your machine:\n'
                  f'Every test creates {test_confs_results[0][1]} instances ({test_confs_results[0][2]} fields) of {model.__name__} model,\n'
                  f'Here is the results:\n'
                  f'\n')
            
            min_time = min(list(map(lambda result: result[0], test_confs_results)))
            min_conf_text = ''
            for i, test_confs_result in enumerate(test_confs_results):
                test_conf_text = ", ".join([f"{k} = {v}" for k, v in test_confs[i].items()])
                print(f'Configuration: {test_conf_text} took {test_confs_result[0]}s')
                if test_confs_result[0] == min_time:
                    min_conf_text = test_conf_text
            print(f'\n\n'
                  f'The best configuration: {min_conf_text}\n')
        
        count = 1000
        model = TaskChallenge
        run_test(count, model)
    except BaseException as ex:
        print(ex)
        have_exception = True
    
    clean_db_after_test(connection_pool, prefix)
    return have_exception


def run_tests():
    connection_pool = redis.ConnectionPool(
        host=os.environ['REDIS_HOST'],
        port=os.environ['REDIS_PORT'],
        db=0,
        decode_responses=True
    )
    tests = [
        basic_test,
        auto_reg_test,
        no_connection_pool_test,
        choices_test,
        order_test,
        filter_test,
        functions_like_defaults_test,
        redis_foreign_key_test,
        update_test,
        delete_test,
        save_consistency_test,
        meta_ttl_test,
        use_keys_test,
        list_test,
        dict_test,
        non_blocking_test,
        foreign_key_test,
        many_to_many_test,
        save_override_test,
        performance_test,
    ]
    results = []
    started_in = datetime.datetime.now()
    print('STARTING TESTS\n')
    for i, test in enumerate(tests):
        print(f'Starting {int(i + 1)} test: {test.__name__.replace("_", " ")}')
        test_started_in = datetime.datetime.now()
        result = not test(connection_pool, test.__name__)
        test_ended_in = datetime.datetime.now()
        test_time = (test_ended_in - test_started_in).total_seconds()
        print(f'{result = } / {test_time}s\n')
        results.append(result)
    ended_in = datetime.datetime.now()
    time = (ended_in - started_in).total_seconds()
    success_message = 'SUCCESS' if all(results) else 'FAILED'
    print('\n'
          f'{success_message}!\n')
    results_success_count = 0
    for i, result in enumerate(results):
        result_message = 'SUCCESS' if result else 'FAILED'
        print(f'Test {(i + 1)}/{len(results)}: {result_message} ({tests[i].__name__.replace("_", " ")})')
        if result:
            results_success_count += 1
    print(f'\n'
          f'{results_success_count} / {len(results)} tests ran successfully\n'
          f'All tests completed in {time}s\n')
    
    return all(results)


if __name__ == '__main__':
    results = run_tests()
    if not results:
        sys.exit(1)

Output

STARTING TESTS

Starting 1 test: basic test
result = True / 0.017655s

Starting 2 test: auto reg test
result = True / 0.002688s

Starting 3 test: no connection pool test
2021-09-17 13:33:42.915213 - RedisRoot: No connection_pool provided, trying default config...
result = True / 0.003571s

Starting 4 test: choices test
result = True / 0.001307s

Starting 5 test: order test
result = True / 0.005999s

Starting 6 test: filter test
result = True / 0.019395s

Starting 7 test: functions like defaults test
result = True / 0.003462s

Starting 8 test: redis foreign key test
result = True / 0.006697s

Starting 9 test: update test
result = True / 0.003751s

Starting 10 test: delete test
result = True / 0.006936s

Starting 11 test: save consistency test
result = True / 6.009583s

Starting 12 test: meta ttl test
result = True / 6.010543s

Starting 13 test: use keys test
Keys usage gives +32.47% efficiency
result = True / 1.348907s

Starting 14 test: list test
result = True / 0.002379s

Starting 15 test: dict test
result = True / 0.002316s

Starting 16 test: non blocking test
Non blocking gives +195.91% efficiency
result = True / 13.0517s

Starting 17 test: foreign key test
result = True / 0.00598s

Starting 18 test: many to many test
result = True / 0.033524s

Starting 19 test: save override test
result = True / 0.003881s

Starting 20 test: performance test



Performance test results on your machine:
Every test creates 1000 instances (6000 fields) of TaskChallenge model,
Here is the results:


Configuration: use_keys = False, use_non_blocking = False took 40.92255s
Configuration: use_keys = False, use_non_blocking = True took 0.234719s
Configuration: use_keys = True, use_non_blocking = False took 31.373307s
Configuration: use_keys = True, use_non_blocking = True took 0.242484s


The best configuration: use_keys = False, use_non_blocking = True

result = True / 73.106303s


SUCCESS!

Test 1/20: SUCCESS (basic test)
Test 2/20: SUCCESS (auto reg test)
Test 3/20: SUCCESS (no connection pool test)
Test 4/20: SUCCESS (choices test)
Test 5/20: SUCCESS (order test)
Test 6/20: SUCCESS (filter test)
Test 7/20: SUCCESS (functions like defaults test)
Test 8/20: SUCCESS (redis foreign key test)
Test 9/20: SUCCESS (update test)
Test 10/20: SUCCESS (delete test)
Test 11/20: SUCCESS (save consistency test)
Test 12/20: SUCCESS (meta ttl test)
Test 13/20: SUCCESS (use keys test)
Test 14/20: SUCCESS (list test)
Test 15/20: SUCCESS (dict test)
Test 16/20: SUCCESS (non blocking test)
Test 17/20: SUCCESS (foreign key test)
Test 18/20: SUCCESS (many to many test)
Test 19/20: SUCCESS (save override test)
Test 20/20: SUCCESS (performance test)

20 / 20 tests ran successfully
All tests completed in 99.646971s