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

```python
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:

```bash
*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:

| First byte | Meaning | Example |
| --- | --- | --- |
| + | Simple string | +OK\r\n |
| - | Error | -ERR unknown command 'FLY'\r\n |
| : | Integer | :42\r\n |
| $ | Bulk string | $3\r\nada\r\n |
| * | Array | *2\r\n... |

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

```python
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`.

```python
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.

```python
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:

```python
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:

```python
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:

```python
# 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.

```python
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:

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

```markdown
> 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:

```sql
----------------------------------------------------------------------
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](https://github.com/alexsamod/blogs)

---

*Published via [ZyVOP](https://zyvop.com/i-built-my-own-redis-clone-in-python-to-understand-how-it-works-k5281?utm_source=hashnode&utm_medium=crosspost&utm_campaign=syndication) — Write once in Markdown, auto-backup to GitHub, and syndicate to Dev.to, Medium & Hashnode in 1 click.*