First time with Redis | Linux, Python 3, and Redis

Background

Without going in to much detail, I didn’t know very much about Redis or how it worked, so I wanted to test it out with Python. Hopefully this script could be of some use for anyone starting out, wanting to play with new tools.
Written with Arch Linux in mind.
If you’re not using PyCharm or a similar IDE, you’re nuts.

Setup and Use

Install packages

$ pacman -S redis python python-pip

Install Redis python libraries

$ pip install python-redis pprint

Configure Redis

Redis will bind to 0.0.0.0 by default, so it’ll be listening for anyone trying to connect to it on port 6379. If you’re just playing around with it on your local machine, it is wise to change the binding to localhost. Use sed:

$ sudo sed -i 's/bind 0.0.0.0/bind 127.0.0.1/g' /etc/redis.conf

Verify it by grepping the config file:

$ grep bind /etc/redis.conf | grep -v "#" bind 127.0.0.1

Start Redis

$ sudo systemctl start redis && sudo systemctl status redis

Python script for interacting with Redis

Some of the structure of the script was inspired by a few templates I found online.
I aimed to make in minimal, but also have some basic exception handling, and testing options to toggle via some booleans

#!/usr/bin/env python3
# Python 3.6
# pip install redis pprint
####################################################################################################
import redis
from os import _exit
from sys import exit
from pprint import pprint
from traceback import print_exc
from datetime import datetime
# from time import sleep

__author__ = "Me"
__copyright__ = "N/A"
__credits__ = ["Me", "The Flying Spaghetti Monster"]
__license__ = "N/A"
__version__ = "0.0.1"
__maintainer__ = __author__
__email__ = "N/A"
__status__ = "Testing"

s_info = "[INFO] "
s_err = "[ERROR] "

# Testing vars
dictionary = {"person1": "Alice",
              "person2": "Bob",
              "person3": "Charlie"}
b_flush_db = True
b_dbsize = True
b_info = False
b_set = True
b_get = True
####################################################################################################


def redis_flushdb(pool):
    redis.Redis(connection_pool=pool).flushdb()
    print(s_info + "DB has been flushed")


def redis_dbsize(pool):
    print(s_info + "DB Size: " + str(redis.Redis(connection_pool=pool).dbsize()))


def redis_info(pool):
    pprint(redis.Redis(connection_pool=pool).info())


def redis_set_value(pool, key, value):
    redis.Redis(connection_pool=pool).set(key, value)
    print(s_info + "Set: \"" + key + "\":\"" + value + "\"")


def redis_get_value(pool, key):
    return str(redis.Redis(connection_pool=pool).get(key))


def main():
    str_start_time = str(datetime.today())

    # CONNECT
    print(s_info + "Connecting to Redis")
    # Create a connection pool to share amongst all operations
    pool = redis.ConnectionPool(host='localhost', port=6379, db=0, socket_timeout=5)

    print(s_info + "Connection Pool PID: " + str(pool.pid))

    t_redis_version = redis.VERSION
    s_version = ""
    for version_digit in t_redis_version:
        s_version = s_version + str(version_digit) + "."
    print(s_info + "Redis Version: " + str(s_version)[:6])
    print(s_info + "Dictionary: ", end='')  # 'end' trims the newline before pprint
    pprint(dictionary)

    # See if testing booleans are true. If so, execute per boolean
    # GET DBSIZE
    if b_dbsize:
        redis_dbsize(pool)
    # FLUSH DB
    if b_flush_db:
        redis_flushdb(pool)
    # GET DBSIZE
    if b_dbsize:
        redis_dbsize(pool)
    # INFO
    if b_info:
        redis_info(pool)
    # SET KEY/VALUE
    if b_set:
        print(s_info + "Setting keys/values...")
        for k, v in dictionary.items():  # Loop through dictionary keys/values
            redis_set_value(pool, k, v)
    # GET VALUE
    if b_get:
        print(s_info + "Getting keys/values...")
        for k in dictionary.keys():
            print(s_info + "Looking for Key: " + str(k))
            print(s_info + "From Redis: " + str(redis_get_value(pool, k))[1:])  # Trim byte char

    # sleep(30)
    print(s_info + "Disconnecting from Redis")
    pool.disconnect()
    print(s_info + "Start time: " + str_start_time)
    print(s_info + "End time:   " + str(datetime.today()))


if __name__ == '__main__':
    try:
        main()
        exit(0)
    except KeyboardInterrupt as e:  # Ctrl-C
        raise e
    except SystemExit as e:  # sys.exit()
        raise e
    except redis.TimeoutError:
        print(s_err + "Connection attempt timed out")
    except redis.ConnectionError:
        print(s_err + "Connection Error")
    except redis.AuthenticationError:
        print(s_err + "Authentication Error")
    except redis.InvalidResponse:
        print(s_err + "Invalid Response from Redis")
    except Exception as e:
        print(s_err + "UNEXPECTED EXCEPTION" + "\n" + "Message: " + str(e))
        print_exc()
        _exit(1)

Console Output

/usr/bin/python3.6 /home/me/PycharmProjects/python-redis/redis-test.py
[INFO] Start time: 2017-08-22 14:17:53.967121
[INFO] Connecting to Redis
[INFO] Connection Pool PID: 22237
[INFO] Redis Version: (2, 10, 6)
[INFO] Dictionary: {'person1': 'Alice', 'person2': 'Bob', 'person3': 'Charlie'} [INFO] DB Size: 3
[INFO] DB has been flushed
[INFO] DB Size: 0
[INFO] Setting keys/values...
[INFO] Set "person1":"Alice"
[INFO] Set "person2":"Bob"
[INFO] Set "person3":"Charlie"
[INFO] Getting keys/values...
[INFO] Looking for Key: person1
[INFO] From Redis: 'Alice'
[INFO] Looking for Key: person2
[INFO] From Redis: 'Bob'
[INFO] Looking for Key: person3
[INFO] From Redis: 'Charlie'
[INFO] Disconnecting from Redis
[INFO] Start time: 2017-08-22 14:17:53.967121
[INFO] End time: 2017-08-22 14:17:53.970050 Process finished with exit code 0

Misc

Want to prove the communications are occurring through the connection pool, as it should? Uncomment the two lines for sleep in the above Python script, and…

$ netstat -plunta | grep python

tcp 0 0 127.0.0.1:40966 127.0.0.1:43913 CLOSE_WAIT 22237/python3.6 tcp 0 0 127.0.0.1:41668 127.0.0.1:36263 ESTABLISHED 22237/python3.6