I built my own Redis clone in Python to understand how it works

I use Redis all the time, but for years it was a black box. I typed SET and got OK back, and I never thought about what happened in between. So I built a small clone to find out.

The result is about 340 lines of Python with no dependencies, just the standard library. It supports a useful slice of Redis: strings, key expiry, counters, and pattern matching on keys. It speaks the same wire protocol as the real thing. It comes with a test suite of 43 tests that talk to the running server over real TCP connections, and every code block in this post comes straight from the file those tests run against.

Here is what it supports:

PING, ECHO, SET (with EX and PX), GET, DEL, EXISTS, INCR, DECR, EXPIRE, TTL, KEYS, DBSIZE, FLUSHALL

The plan

A Redis server really does four things:

1. Accept TCP connections

2. Parse commands out of the bytes the client sends

3. Run each command against an in-memory dictionary

4. Encode the answer and send it back

Most of the interesting parts hide in steps 2 and 3. Everything lives in one file, which starts with imports from the standard library:

import argparse
import asyncio
import fnmatch
import random
import re
import time

Let's start with the protocol.

Step 1: Learn to speak RESP

Redis clients and servers talk RESP, the Redis serialization protocol. It is far simpler than I expected. When you run SET name ada, the client sends this over the wire:

*3\r\n$3\r\nSET\r\n$4\r\nname\r\n$3\r\nada\r\n

Read it piece by piece. *3 means "an array of 3 items". Each item is a bulk string, written as $ followed by its length in bytes, then the bytes themselves. So $3\r\nSET\r\n is the 3-byte string SET.

Replies use the same idea with a few more type markers:

A missing value is a bulk string with length -1: $-1\r\n. Encoding all of this takes five tiny functions:

def simple(text):
    return b"+" + text.encode() + b"\r\n"


def error(text):
    # Replies are framed by \r\n, so never let one leak into an error message.
    text = text.replace("\r", " ").replace("\n", " ")
    return b"-" + text.encode() + b"\r\n"


def integer(n):
    return b":" + str(n).encode() + b"\r\n"


def bulk(value):
    if value is None:
        return b"$-1\r\n"
    return b"$" + str(len(value)).encode() + b"\r\n" + value + b"\r\n"


def array(encoded_items):
    return b"*" + str(len(encoded_items)).encode() + b"\r\n" + b"".join(encoded_items)

Notice the comment in error(). Replies are framed by \r\n, so if a client could sneak a newline into an error message, it could corrupt the stream. I added a test for exactly that.

Parsing is the harder direction, because bytes arrive in whatever chunks the network feels like. A command can show up split across several packets, or several commands can arrive in one. The trick is to never guess: read a line, and if it says "bulk string of 4 bytes", read exactly 4 bytes plus the trailing \r\n.

MAX_ARGS = 1_000_000


MAX_BULK = 64 * 1024 * 1024


class ProtocolError(Exception):
    pass


async def read_line(reader):
    """Read one line. Returns None if the client disconnected."""
    try:
        line = await reader.readline()
    except ValueError:
        raise ProtocolError("line too long")
    if not line.endswith(b"\n"):
        return None
    return line.rstrip(b"\r\n")


def parse_length(raw, what):
    try:
        return int(raw)
    except ValueError:
        raise ProtocolError(f"invalid {what} length")


async def read_command(reader):
    """Read one command as a list of byte strings, or None if the client left."""
    line = await read_line(reader)
    if line is None:
        return None
    if not line.startswith(b"*"):
        # Inline command, the kind you type into telnet or nc: PING
        return line.split()

    count = parse_length(line[1:], "multibulk")
    if count > MAX_ARGS:
        raise ProtocolError("invalid multibulk length")
    args = []
    for _ in range(max(count, 0)):
        header = await read_line(reader)
        if header is None:
            return None
        if not header.startswith(b"$"):
            raise ProtocolError("expected '$', got " + repr(header[:1].decode("latin-1")))
        length = parse_length(header[1:], "bulk")
        if length < 0 or length > MAX_BULK:
            raise ProtocolError("invalid bulk length")
        try:
            data = await reader.readexactly(length + 2)
        except asyncio.IncompleteReadError:
            return None
        if data[-2:] != b"\r\n":
            raise ProtocolError("bulk string not terminated by CRLF")
        args.append(data[:-2])
    return args

Two details worth pointing out. Because values are length-prefixed rather than delimited, they can contain anything, including \r\n and null bytes. That is what "binary safe" means, and it falls out of the design for free. Second, if a line does not start with *, I treat it as an inline command. That is what lets you type PING into nc by hand.

Step 2: The store

The data itself is just two dictionaries: one for values and one for expiry times.

Expiry is the interesting part. A key with a TTL has to disappear on time, but you do not want a timer per key. Redis uses two strategies together, and so does this clone. The first is lazy expiration: whenever a key is touched, check its deadline first, and if it has passed, delete it and act as if it never existed. The second I will cover in a moment.

class Store:
    """Keys and values, plus an optional expiry time for each key."""

    def __init__(self):
        self.data = {}
        self.expires = {}  # key -> unix timestamp

    def purge_if_expired(self, key):
        """Lazy expiration: a key is removed the moment someone touches it too late."""
        deadline = self.expires.get(key)
        if deadline is not None and time.time() >= deadline:
            self.data.pop(key, None)
            del self.expires[key]
            return True
        return False

    def get(self, key):
        self.purge_if_expired(key)
        return self.data.get(key)

    def set(self, key, value, ttl=None):
        self.data[key] = value
        if ttl is None:
            self.expires.pop(key, None)  # a plain SET clears any old TTL
        else:
            self.expires[key] = time.time() + ttl

    def delete(self, key):
        self.purge_if_expired(key)
        self.expires.pop(key, None)
        return self.data.pop(key, None) is not None

    def expire(self, key, seconds):
        self.purge_if_expired(key)
        if key not in self.data:
            return False
        self.expires[key] = time.time() + seconds
        return True

    def ttl(self, key):
        """Seconds left, -1 if the key has no expiry, -2 if it does not exist."""
        self.purge_if_expired(key)
        if key not in self.data:
            return -2
        deadline = self.expires.get(key)
        if deadline is None:
            return -1
        return max(0, round(deadline - time.time()))

A few behaviors here match real Redis on purpose. A plain SET clears any existing TTL. TTL returns -1 for a key with no expiry and -2 for a key that does not exist.

Step 3: The commands

Each command is a function that takes the store and its arguments and returns an already-encoded reply. I start with the helpers, plus SET, which is the most involved command because of its options:

class CommandError(Exception):
    pass


def parse_int(raw):
    # int() alone would accept things Redis rejects, like b" 5" or b"1_0".
    if not re.fullmatch(rb"-?[0-9]+", raw):
        raise CommandError("ERR value is not an integer or out of range")
    n = int(raw)
    if not -(2 ** 63) <= n < 2 ** 63:
        raise CommandError("ERR value is not an integer or out of range")
    return n


def cmd_ping(store, args):
    return bulk(args[0]) if args else simple("PONG")


def cmd_echo(store, args):
    return bulk(args[0])


def cmd_set(store, args):
    key, value, *options = args
    ttl = None
    i = 0
    while i < len(options):
        option = options[i].upper()
        if option in (b"EX", b"PX") and i + 1 < len(options):
            amount = parse_int(options[i + 1])
            if amount <= 0:
                raise CommandError("ERR invalid expire time in 'set' command")
            ttl = amount if option == b"EX" else amount / 1000
            i += 2
        else:
            raise CommandError("ERR syntax error")
    store.set(key, value, ttl)
    return simple("OK")

parse_int looks over-engineered, but Python's int() is too forgiving. It happily accepts b" 5" and even b"1_0" (which it reads as 10), while Redis rejects both. A regular expression keeps the behavior strict.

The rest of the commands are short:

def cmd_get(store, args):
    return bulk(store.get(args[0]))


def cmd_del(store, args):
    return integer(sum(store.delete(key) for key in args))


def cmd_exists(store, args):
    return integer(sum(store.get(key) is not None for key in args))


def add_to_counter(store, key, delta):
    raw = store.get(key)
    result = (0 if raw is None else parse_int(raw)) + delta
    if not -(2 ** 63) <= result < 2 ** 63:
        raise CommandError("ERR increment or decrement would overflow")
    store.data[key] = str(result).encode()  # writing to data directly keeps the TTL
    return integer(result)


def cmd_incr(store, args):
    return add_to_counter(store, args[0], 1)


def cmd_decr(store, args):
    return add_to_counter(store, args[0], -1)


def cmd_expire(store, args):
    seconds = parse_int(args[1])
    if seconds <= 0:
        return integer(int(store.delete(args[0])))
    return integer(int(store.expire(args[0], seconds)))


def cmd_ttl(store, args):
    return integer(store.ttl(args[0]))


def cmd_keys(store, args):
    pattern = args[0].decode("latin-1")
    matches = []
    for key in list(store.data):
        if store.purge_if_expired(key):
            continue
        if fnmatch.fnmatchcase(key.decode("latin-1"), pattern):
            matches.append(bulk(key))
    return array(matches)


def cmd_dbsize(store, args):
    return integer(len(store.data))


def cmd_flushall(store, args):
    store.data.clear()
    store.expires.clear()
    return simple("OK")


def cmd_command(store, args):
    # Some versions of redis-cli send this when they connect. An empty answer is fine.
    return array([])

Look at add_to_counter. INCR writes straight into store.data instead of calling store.set(), because set() would clear the TTL, and real Redis keeps the TTL on INCR. The test suite has a dedicated test for this.

Finally, a dispatch table maps command names to functions and their allowed argument counts, so arity checking happens in one place:

# name -> (function, minimum arguments, maximum arguments or -1 for no limit)
COMMANDS = {
    "PING": (cmd_ping, 0, 1),
    "ECHO": (cmd_echo, 1, 1),
    "SET": (cmd_set, 2, -1),
    "GET": (cmd_get, 1, 1),
    "DEL": (cmd_del, 1, -1),
    "EXISTS": (cmd_exists, 1, -1),
    "INCR": (cmd_incr, 1, 1),
    "DECR": (cmd_decr, 1, 1),
    "EXPIRE": (cmd_expire, 2, 2),
    "TTL": (cmd_ttl, 1, 1),
    "KEYS": (cmd_keys, 1, 1),
    "DBSIZE": (cmd_dbsize, 0, 0),
    "FLUSHALL": (cmd_flushall, 0, 0),
    "COMMAND": (cmd_command, 0, -1),
}


def execute(store, args):
    name = args[0].decode("latin-1").upper()
    entry = COMMANDS.get(name)
    if entry is None:
        return error(f"ERR unknown command '{name}'")
    handler, min_args, max_args = entry
    count = len(args) - 1
    if count < min_args or (max_args != -1 and count > max_args):
        return error(f"ERR wrong number of arguments for '{name.lower()}' command")
    try:
        return handler(store, args[1:])
    except CommandError as exc:
        return error(str(exc))

Step 4: The server

This is where asyncio earns its keep. Each client connection gets its own coroutine that loops: read a command, run it, write the reply.

async def handle_client(reader, writer, store):
    try:
        while True:
            args = await read_command(reader)
            if args is None:
                break
            if args:
                writer.write(execute(store, args))
                await writer.drain()
    except ProtocolError as exc:
        writer.write(error(f"ERR Protocol error: {exc}"))
    except ConnectionError:
        pass
    finally:
        writer.close()


async def expire_sweeper(store, interval=0.1, sample_size=20):
    """Active expiry: on every tick, check a random sample of keys that have a TTL."""
    while True:
        await asyncio.sleep(interval)
        sample = random.sample(list(store.expires), min(sample_size, len(store.expires)))
        for key in sample:
            store.purge_if_expired(key)


async def serve(host="127.0.0.1", port=6380):
    store = Store()
    server = await asyncio.start_server(
        lambda reader, writer: handle_client(reader, writer, store), host, port
    )
    sweeper = asyncio.create_task(expire_sweeper(store))
    addresses = ", ".join(str(sock.getsockname()) for sock in server.sockets)
    print(f"Listening on {addresses}", flush=True)
    async with server:
        await server.serve_forever()
    sweeper.cancel()

And the entry point:

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="A tiny Redis clone")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=6380)
    options = parser.parse_args()
    try:
        asyncio.run(serve(options.host, options.port))
    except KeyboardInterrupt:
        pass

There is a design choice here that is easy to miss. The whole server runs on a single thread, and execute() is a plain synchronous function with no await inside it. That means once a command starts running, nothing else can run until it finishes. Two clients can never be halfway through an INCR at the same time, so counters are atomic without a single lock. Redis executes commands one at a time on a single thread for the same reason, and it was satisfying to see the same idea work in a few lines of Python.

The test suite checks it: 20 threads each send 100 INCR commands to the same key, and the final value comes out to exactly 2000, with no lost updates.

The expire_sweeper is the second expiry strategy, called active expiration. Lazy expiry alone has a flaw: a key that nobody ever reads again would sit in memory forever. So a background task wakes up ten times a second, picks a random sample of keys that have a TTL, and removes the expired ones. Real Redis does something similar, sampling keys with a TTL several times per second.

Try it

Start the server:

python redis_clone.py

It listens on port 6380 by default, so it will not clash with a real Redis on 6379. You can connect with nc localhost 6380 and type commands by hand. If you have redis-cli installed, try pointing it at port 6380 as well. Here is a real session (I trimmed the trailing \r\n from each reply to keep it readable):

> PING
+PONG
> SET name ada
+OK
> GET name
$3
ada
> INCR visits
:1
> INCR visits
:2
> SET session abc EX 30
+OK
> TTL session
:30
> GET missing
$-1
> INCR name
-ERR value is not an integer or out of range
> FLY
-ERR unknown command 'FLY'

Pipelining works too, with no extra code. Because the server reads commands in a loop, a client can send several commands in one packet and read the replies back in order.

Testing it

I did not want to claim this works based on a few manual checks, so the test suite starts the real server as a subprocess and talks to it over TCP using a tiny RESP client that I wrote separately from the server code. Here is what it covers:

• Every command, including edge cases such as DEL a a and EXISTS a a

• Expiry with EX and PX, plus the background sweeper, which is checked using DBSIZE because it never looks at individual keys

• Strict integer parsing, overflow, and INCR keeping its TTL

• Binary-safe values, Unicode keys, and a 1 MB value

• Pipelined commands, commands split across packets, and inline commands

• Malformed input, which should produce an error instead of a crash

• Clients that disconnect halfway through a command

• 200 simultaneous connections and the 20-thread counter race

Run it with python test_redis_clone.py. The tail of the output looks like this:

----------------------------------------------------------------------
Ran 43 tests in 3.102s

OK

What I learned

The protocol is the easy part to love. RESP is small enough to keep in your head. Once you can read and write it, you can poke at a real Redis with nc and understand what the server is doing.

Single-threaded does not mean slow or simple, it means no locks. Having one thread run commands to completion removed a whole category of bugs.

Expiry is a trade-off. Lazy deletion is cheap but leaks memory for forgotten keys. Active deletion fixes that but costs CPU. Using both is the compromise.

The edges take most of the effort. The core commands are a few lines each. The care goes into INCR and TTLs, strict number parsing, and commands that arrive in pieces.

What is missing

This is a learning project, not a replacement for Redis. It has no persistence, so everything disappears when the process stops. It only supports strings. Real Redis also has lists, hashes, sets, sorted sets, pub/sub, transactions, replication, and memory limits. Some shortcuts are visible in the code too: the sweeper copies the list of TTL keys on every tick, which is fine for a toy and would be a problem with millions of keys, and there is no authentication.

If you want to extend it, adding a LPUSH and LRANGE pair for lists is a good next step, followed by saving the dictionary to disk on shutdown.

View the complete Redis clone on GitHub https://github.com/alexsamod/blogs