Build Your Own Redis: Core Server
Build a Redis server from scratch in Python that speaks the real Redis protocol.
Introduction
30 Second Summary
Every app you use relies on lightning-fast data storage happening behind the scenes. Redis is one of the most popular tools powering that speed, keeping everything in memory rather than on disk.
In this project, you will build a working Redis server from scratch in Python. By the end, any Redis client can connect to your server and run real commands against it.
What You'll Build
A TCP server that implements the Redis Serialization Protocol (RESP2), accepts concurrent client connections, and stores key-value data with optional time-based expiry.
By the end of this project, you'll have:
- A TCP server accepting concurrent connections on port 6379.
- A RESP2 protocol parser that encodes and decodes Redis wire format messages.
- A command router dispatching PING, ECHO, SET, and GET to handler functions.
- An in-memory key-value store with time-based key expiry using EX and PX options.
- Secret Mission: Implement EXISTS, DEL, and KEYS commands with glob pattern matching.
Do I need networking experience?
Nope. This project assumes basic Python knowledge (functions, dictionaries, loops). You'll learn networking and async I/O concepts as you build.
Set Up the Project and Test Client
Building a Redis server means speaking a network protocol. Before you write the server itself, you need a reliable way to test it at every milestone.
In this step, you'll write a custom test client using Python's socket module. This client will attempt to connect to the server's port and report what it finds.
In this step, get ready to:
- Write a reusable RESP command function for sending commands to the server.
- Create a connection test that confirms the client can reach port 6379.
Write the test client
Your test client is a small Python script that opens a TCP connection to localhost:6379 (the default Redis port) and tries to communicate. You'll use it after every server change to verify things work.
Why not use redis-cli?
Installing redis-cli on Windows requires WSL or a third-party binary. A raw socket client has zero friction and teaches you both sides of the protocol.
By encoding commands yourself, you prove you understand the wire format your server must speak.
- Switch back to test_client.py in VS Code and add the imports and command utility function:
import socket
import time
def send_resp_command(sock, *args):
# Build a RESP array header with the argument count
cmd = f"*{len(args)}\r\n"
for arg in args:
# Encode each argument as a RESP bulk string
arg_str = str(arg)
cmd += f"${len(arg_str)}\r\n{arg_str}\r\n"
# Send the full command and return the server's response
sock.sendall(cmd.encode())
return sock.recv(4096).decode()What does this code do?
The send_resp_command function encodes any command into the RESP wire format that Redis speaks. RESP stands for Redis Serialization Protocol.
- *{count}\r\n is the RESP array header, announcing how many elements follow.
- ${length}\r\n{data}\r\n is a RESP bulk string, wrapping each argument with its byte length.
- sock.sendall() sends the encoded bytes over the TCP connection, then sock.recv(4096) reads back up to 4096 bytes of the server's response.
The time import is for expiry tests you'll write in a later step.
You won't see a change yet. The next chunk adds the entry point that connects to the server.
- Below send_resp_command, add the main() function and entry point:
def main():
# Create a TCP socket and connect to the Redis server port
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("localhost", 6379))
# Try to receive any data and print it
print(repr(sock.recv(4096).decode()))
sock.close()
if __name__ == "__main__":
main()What does this code do?
- socket.AF_INET specifies IPv4, and socket.SOCK_STREAM specifies TCP (a reliable, ordered byte stream).
- sock.connect(("localhost", 6379)) attempts to reach port 6379 on your own machine.
- repr() shows the raw string with escape characters visible, so you can see protocol formatting like \r\n.
- if __name__ == "__main__": ensures main() only runs when you execute the file directly (not when imported).
- Save test_client.py.
- Run the test client to verify the connection attempt:
python test_client.pyYou should see a ConnectionRefusedError in your terminal. This confirms your client is correctly trying to reach port 6379, but no server is listening yet.
Seeing a different error?
- If you see a SyntaxError, check that your f-strings use matching curly braces and that \r\n sequences are inside the string quotes.
- If the script runs with no output at all, make sure you included the if __name__ == "__main__": block at the bottom of the file.
- If you see NameError: name 'socket' is not defined, confirm that import socket is the very first line of your file.
Help me debug my test_client.py connection error.
✔️ Awesome, I've got everything!
Great. Double check you've saved test_client.py and confirmed the ConnectionRefusedError appears when you run it.
ⓧ I'd like to double check the full code
import socket
import time
def send_resp_command(sock, *args):
# Build a RESP array header with the argument count
cmd = f"*{len(args)}\r\n"
for arg in args:
# Encode each argument as a RESP bulk string
arg_str = str(arg)
cmd += f"${len(arg_str)}\r\n{arg_str}\r\n"
# Send the full command and return the server's response
sock.sendall(cmd.encode())
return sock.recv(4096).decode()
def main():
# Create a TCP socket and connect to the Redis server port
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("localhost", 6379))
# Try to receive any data and print it
print(repr(sock.recv(4096).decode()))
sock.close()
if __name__ == "__main__":
main()Your test client is ready and waiting for a server to talk to. Next up, you'll build that server and watch it respond to its first connection.
Bind to a Port and Respond to PING
Your test client is ready, but it has nothing to talk to. Running it right now produces a "Connection refused" error because no server exists on port 6379 yet.
In this step, you'll create the server. You'll use Python's asyncio library to bind a TCP server to port 6379 and respond to every incoming connection with the Redis PING response.
In this step, get ready to:
- Write an async connection handler that responds with +PONG\r\n.
- Start a TCP server that listens on port 6379.
- Test the server by running your test client against it.
Write the connection handler
Every time a client connects to your server, the server needs a function to handle that connection. This function will read whatever the client sends, ignore it, and reply with a hardcoded PONG response.
- In VS Code, open server.py (it should be empty) and add the connection handler by pasting this code:
import asyncio
async def handle_client(reader, writer):
await reader.read(1024)
writer.write(b"+PONG\r\n")
await writer.drain()
writer.close()What does this code do?
- reader.read(1024) reads whatever the client sends, up to 1024 bytes. You don't care what it says yet.
- writer.write(b"+PONG\r\n") sends back the RESP simple string format for "PONG". The + prefix tells Redis clients this is a simple string response.
- writer.drain() ensures the bytes are actually flushed to the network before continuing.
- writer.close() ends the connection after the single response.
You won't see a change yet. The next chunk adds the entry point that actually starts the server.
Add the server entry point
A handler function on its own doesn't do anything. You need a main() function that binds the server to a port and tells it to listen for connections forever.
- Add the following code below your handle_client function in server.py:
async def main():
server = await asyncio.start_server(handle_client, "localhost", 6379)
addr = server.sockets[0].getsockname()
print(f"Redis server listening on {addr[0]}:{addr[1]}")
async with server:
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())What does this code do?
- asyncio.start_server(handle_client, "localhost", 6379) binds a TCP server to localhost on port 6379. Every new connection triggers handle_client automatically.
- server.sockets[0].getsockname() retrieves the actual address the server bound to, so you can confirm it in the terminal output.
- async with server: keeps the server alive. serve_forever() runs the event loop until you stop the process with Ctrl+C.
- asyncio.run(main()) is the standard way to launch an async program in Python.
- Save server.py.
- Run the server by entering this command in your terminal:
python server.pyYou should see Redis server listening on localhost:6379 printed in the terminal. The cursor won't return to a prompt because the server is running continuously.
Server not starting?
- If you see an error about the address already being in use, another process is occupying port 6379. Close any other server or Redis instance running on that port and try again.
- If you see a SyntaxError, double-check that handle_client and main both use the async def keyword, not just def.
help me debug my server startup error
Test the server with your client
Your server is running in one terminal. Now you'll connect to it from a second terminal using the test client you wrote in Step 1.
- Open a second terminal window. Keep the first terminal (with the server) running.
- In the new terminal, run the test client:
python test_client.pyYou should see '+PONG\r\n' printed in the output. The server responds with PONG no matter what the client sends.
Why does the output look like that?
Your test client uses repr() to print the response, which shows the raw string including escape characters. \r\n is the carriage return + newline that the RESP protocol requires at the end of every message.
Still seeing Connection refused?
- Make sure the first terminal still shows Redis server listening on localhost:6379. If you accidentally closed it, run python server.py again.
- Make sure you're running python test_client.py from the redis-server folder. If your terminal opened somewhere else, navigate to the project folder first.
help me figure out why my test client can't connect to the server
✔️ Awesome, I've got everything!
Great. Double check you've saved server.py and confirmed the PONG response.
ⓧ I'd like to double check the full code
import asyncio
async def handle_client(reader, writer):
await reader.read(1024)
writer.write(b"+PONG\r\n")
await writer.drain()
writer.close()
async def main():
server = await asyncio.start_server(handle_client, "localhost", 6379)
addr = server.sockets[0].getsockname()
print(f"Redis server listening on {addr[0]}:{addr[1]}")
async with server:
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())Your server accepts a connection and responds with PONG. Next up, you'll make it handle multiple commands on the same connection and serve multiple clients at once.
Handle Multiple Commands and Concurrent Clients
Your server can accept a connection and respond to one command. But real Redis clients send dozens of commands on a single connection without reconnecting each time.
In this step, you'll upgrade your server to keep connections alive, handle multiple commands per session, and serve multiple clients at the same time without any blocking.
In this step, get ready to:
- Add a loop to keep connections alive across multiple commands.
- Handle client disconnections gracefully with error handling.
- Verify multiple simultaneous clients can interact with the server.
Add a persistent connection loop to the server
Right now, your handle_client function reads one chunk of data, writes +PONG\r\n, and immediately closes the connection. A real server keeps the connection open so a client can send as many commands as it wants.
- In server.py, replace the entire handle_client function with this updated version:
async def handle_client(reader, writer):
# Log the client's address for debugging
addr = writer.get_extra_info("peername")
print(f"Client connected: {addr}")
try:
while True:
# Read up to 1024 bytes from the client
data = await reader.read(1024)
# Empty bytes means the client disconnected
if not data:
break
# Respond with PONG regardless of what was sent
writer.write(b"+PONG\r\n")
await writer.drain()
except ConnectionResetError:
# Client disconnected abruptly (e.g. killed the process)
pass
finally:
# Always clean up the connection
print(f"Client disconnected: {addr}")
writer.close()
await writer.wait_closed()What does this code do?
The while True loop keeps the connection open. Each iteration reads data from the client and sends back a response. The loop only breaks when reader.read(1024) returns empty bytes, which means the client closed its end of the connection.
The try/except/finally structure handles the case where a client disconnects abruptly (like crashing or closing the terminal). Without it, a ConnectionResetError would crash your server. The finally block ensures the connection is always cleaned up, no matter how the client left.
- Save server.py.
Seeing a syntax error?
- Make sure you replaced the entire handle_client function, including the async def line. Mixing old and new code will cause indentation errors.
- Check that the while True block is indented inside try: and that except and finally are at the same indentation level as try.
Why is my server crashing with indentation errors?
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
import asyncio
async def handle_client(reader, writer):
addr = writer.get_extra_info("peername")
print(f"Client connected: {addr}")
try:
while True:
data = await reader.read(1024)
if not data:
break
writer.write(b"+PONG\r\n")
await writer.drain()
except ConnectionResetError:
pass
finally:
print(f"Client disconnected: {addr}")
writer.close()
await writer.wait_closed()
async def main():
server = await asyncio.start_server(handle_client, "localhost", 6379)
addr = server.sockets[0].getsockname()
print(f"Redis server listening on {addr[0]}:{addr[1]}")
async with server:
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())Update the test client to send multiple commands
Your old test client connected, received one response, and closed. Now that the server keeps connections alive, you need a test client that sends multiple commands on the same connection to prove it works.
- In test_client.py, replace the main() function with this version that sends three PINGs without reconnecting:
def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("localhost", 6379))
# Send three PINGs on the same connection
print("--- Test multiple PINGs on one connection ---")
for i in range(3):
sock.sendall(b"*1\r\n$4\r\nPING\r\n")
print(repr(sock.recv(4096).decode()))
sock.close()What does this code do?
The test sends three PING commands as raw RESP-encoded bytes. The string *1\r\n$4\r\nPING\r\n is the RESP encoding for a single-element array containing the bulk string "PING". This is the exact format that real Redis clients use to send commands.
Each send happens on the same socket connection. If the server closed the connection after the first response (like it used to), the second sendall would fail with an error.
- Save test_client.py.
Getting an error about sendall?
- Make sure you kept the import socket and import time lines at the top of the file. Only the main() function should be replaced.
- Keep the send_resp_command function from the previous step. You'll use it again in a later step.
Help me fix my test_client.py file
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on.
ⓧ I'd like to double check the full code
import socket
import time
def send_resp_command(sock, *args):
cmd = f"*{len(args)}\r\n"
for arg in args:
arg_str = str(arg)
cmd += f"${len(arg_str)}\r\n{arg_str}\r\n"
sock.sendall(cmd.encode())
return sock.recv(4096).decode()
def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("localhost", 6379))
print("--- Test multiple PINGs on one connection ---")
for i in range(3):
sock.sendall(b"*1\r\n$4\r\nPING\r\n")
print(repr(sock.recv(4096).decode()))
sock.close()
if __name__ == "__main__":
main()Test persistent connections and concurrency
Time to prove both features work. You'll run the server, hit it with one client, then hit it with two clients at the same time.
- Start your server by running this command in a terminal:
python server.py- Open a second terminal and run the test client:
python test_client.pyYou should see three +PONG\r\n responses printed, one for each PING sent on the same connection.
- Now test concurrency. Open a third terminal and run python test_client.py in both the second and third terminals at roughly the same time.
Both clients should receive their three PONG responses without waiting for each other.
Why does concurrency work without threads?
When a new client connects, asyncio.start_server spawns a new coroutine for that connection automatically. Each coroutine runs independently on the same event loop. While one coroutine is waiting on reader.read(), others can send and receive data.
This is the same model that real Redis uses. Redis is single-threaded but handles thousands of concurrent clients through an event loop. Your server works the same way.
- Check the terminal where your server is running. You should see connection and disconnection messages for each client.
- Stop the server by pressing Ctrl+C in the server terminal.
Only seeing one PONG response?
- Make sure you saved server.py before starting the server. If you started an old version, stop it with Ctrl+C and run python server.py again.
- If port 6379 is already in use, stop any other process on that port first. On macOS/Linux, try lsof -i :6379 to find it.
My server is not responding to multiple commands
Your server now holds connections open and handles as many clients as you throw at it. Next up, you'll teach it to actually understand what clients are saying by parsing the RESP protocol and implementing the ECHO command.
Parse the RESP Protocol and Implement ECHO
Your server handles multiple clients sending multiple commands on persistent connections. Right now it returns +PONG\r\n no matter what bytes arrive. It has no idea what a client is actually asking for.
Real Redis clients don't send plain text. They encode every command in a binary format called RESP2 (Redis Serialization Protocol). In this step, you'll build a parser that decodes RESP messages into Python values, and encoder functions that format responses back into the wire format. Once both sides speak the same language, you'll implement your first real command: ECHO.
In this step, get ready to:
- Build an async RESP2 protocol parser that decodes arrays, bulk strings, simple strings, and integers.
- Create encoder functions that format server responses into valid RESP bytes.
- Implement a command router with working PING and ECHO commands.
Build the RESP protocol parser
Every RESP message starts with a single-byte type prefix. The prefix tells you what kind of data follows and how to read the rest of the message.
What is RESP2?
RESP2 is the wire format Redis clients and servers use to communicate. Each data type has its own prefix character:
- + = Simple String (e.g., +OK\r\n).
- - = Error (e.g., -ERR unknown command\r\n).
- : = Integer (e.g., :1000\r\n).
- $ = Bulk String with a length prefix (e.g., $5\r\nhello\r\n).
- * = Array with a count prefix (e.g., *2\r\n$4\r\nPING\r\n$5\r\nhello\r\n).
Clients always send commands as RESP arrays of bulk strings. Your parser needs to handle all five types because responses use them too.
- In server.py, add import time after the existing import asyncio line, then add two module-level dictionaries below the imports. These prepare the file for the key-value store you'll build in the next step.
- Below the two dictionaries (and above the existing handle_client function), add the read_resp function by pasting this code:
import asyncio
import time
store = {}
expiry = {}
async def read_resp(reader):
# Read the first line to determine the message type
line = await reader.readline()
if not line:
return None
# First byte is the type prefix, rest is the data
prefix = chr(line[0])
data = line[1:].strip().decode()
if prefix == "+":
# Simple string - return as-is
return data
elif prefix == "-":
# Error - wrap in an Exception
return Exception(data)
elif prefix == ":":
# Integer - convert to int
return int(data)
elif prefix == "$":
# Bulk string - read exactly 'length' bytes plus \r\n
length = int(data)
if length == -1:
return None
content = await reader.readexactly(length + 2)
return content[:length].decode()
elif prefix == "*":
# Array - recursively read 'count' elements
count = int(data)
if count == -1:
return None
elements = []
for _ in range(count):
element = await read_resp(reader)
elements.append(element)
return elements
else:
return dataWhat does this code do?
The read_resp function reads one complete RESP message from the stream. It uses reader.readline() to grab the first line (which always contains the type prefix and metadata).
- For bulk strings ($), it uses reader.readexactly(length + 2) to guarantee it receives the full payload. The + 2 accounts for the trailing \r\n.
- For arrays (*), it calls itself recursively once per element. A command like ECHO hello arrives as an array of two bulk strings, so read_resp returns ["ECHO", "hello"].
Why readexactly instead of read?
The reader.read(n) method may return fewer than n bytes if the data hasn't fully arrived over the network yet. reader.readexactly(n) guarantees exactly n bytes or raises an IncompleteReadError if the connection drops. For a protocol parser, partial reads mean corrupted data.
- Save server.py.
You won't see a change yet. The next chunk adds the encoder functions that format responses back to the client.
Create the response encoders and command router
Your parser can decode incoming RESP messages into Python values. Now you need the reverse: functions that take Python values and encode them back into RESP bytes for the client.
- In server.py, add these three encoder functions between the expiry = {} line and the read_resp function:
def encode_simple_string(s):
# Format: +<string>\r\n
return f"+{s}\r\n".encode()
def encode_error(msg):
# Format: -ERR <message>\r\n
return f"-ERR {msg}\r\n".encode()
def encode_bulk_string(s):
# Format: $<length>\r\n<data>\r\n (or $-1\r\n for None)
if s is None:
return b"$-1\r\n"
encoded = s.encode() if isinstance(s, str) else s
return f"${len(encoded)}\r\n".encode() + encoded + b"\r\n"What does this code do?
- encode_simple_string wraps a string in the +...\r\n format. Used for short status responses like OK and PONG.
- encode_error formats error responses with the -ERR prefix so clients know something went wrong.
- encode_bulk_string handles variable-length data by including the byte count before the payload. Passing None returns the null bulk string $-1\r\n, which is how Redis represents "key not found."
Now you need a function that takes a parsed command and decides which encoder to call.
- Below the read_resp function and above handle_client, add the command router by pasting this code:
def handle_command(command):
# Reject anything that isn't a list (all valid commands are RESP arrays)
if not command or not isinstance(command, list):
return encode_error("invalid command")
# Extract the command name and arguments
cmd = command[0].upper()
args = command[1:]
if cmd == "PING":
return encode_simple_string("PONG")
elif cmd == "ECHO":
if len(args) != 1:
return encode_error("wrong number of arguments for 'echo' command")
return encode_bulk_string(args[0])
else:
return encode_error(f"unknown command '{cmd}'")What does this code do?
The handle_command function is the command router. It extracts the first element as the command name (uppercased for case-insensitive matching) and the rest as arguments.
- PING returns a simple string +PONG\r\n.
- ECHO returns its argument as a bulk string (with a length prefix).
- Unknown commands get an error response, so the client knows what went wrong.
- Save server.py.
You won't see a change yet. The server still uses the old reader.read(1024) logic. The next chunk connects the parser and router to the client handler.
Wire up the server and test
The parser and encoders are ready. Now you'll update handle_client to use them instead of the raw byte approach, then update your test client to send properly encoded RESP commands.
- In server.py, find your existing handle_client function and replace it entirely with this updated version:
async def handle_client(reader, writer):
addr = writer.get_extra_info("peername")
print(f"Client connected: {addr}")
try:
while True:
# Parse the incoming RESP message into a Python list
command = await read_resp(reader)
if command is None:
break
# Route the command and send the response
response = handle_command(command)
writer.write(response)
await writer.drain()
except (ConnectionResetError, asyncio.IncompleteReadError):
pass
finally:
print(f"Client disconnected: {addr}")
writer.close()
await writer.wait_closed()What changed here?
- reader.read(1024) is replaced by await read_resp(reader). Instead of raw bytes, you now get a parsed Python list like ["PING"] or ["ECHO", "hello"].
- The hardcoded b"+PONG\r\n" is replaced by handle_command(command), which returns the correct response bytes for any supported command.
- The except clause now catches asyncio.IncompleteReadError too. This handles clients that disconnect mid-message (before readexactly finishes).
- Save server.py.
Now update the test client to exercise both PING and ECHO through the proper RESP encoding.
- In test_client.py, find the main() function and replace it entirely with this updated version:
def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("localhost", 6379))
print("--- Test PING ---")
print(repr(send_resp_command(sock, "PING")))
print("\n--- Test ECHO ---")
print(repr(send_resp_command(sock, "ECHO", "hello world")))
sock.close()What does this code do?
The send_resp_command function (which already exists at the top of your file) encodes arguments as a RESP array of bulk strings before sending. This is the exact format real Redis clients use.
For example, send_resp_command(sock, "ECHO", "hello world") sends *2\r\n$4\r\nECHO\r\n$11\r\nhello world\r\n over the wire.
- Save test_client.py.
- Start your server by running this in a terminal:
python server.py- Open a second terminal and run your test client:
python test_client.py- You should see +PONG\r\n for the PING test and $11\r\nhello world\r\n for the ECHO test.
Seeing a connection error or garbled output?
- Make sure you stopped the old server (Ctrl+C) and started a fresh one. The old server doesn't have the parser code.
- Check that the encoder functions are placed ABOVE read_resp in the file, and handle_command is ABOVE handle_client. Python needs functions defined before they're called.
- If you see an IndentationError, make sure each function is at the top level (no extra indentation) and uses consistent spacing.
Help me debug my RESP parser output
✔️ Awesome, I've got everything!
Great. Double check you've saved both files, then stop the server with Ctrl+C before moving on.
ⓧ I'd like to double check the full code
Here is the complete server.py at this point:
import asyncio
import time
store = {}
expiry = {}
def encode_simple_string(s):
return f"+{s}\r\n".encode()
def encode_error(msg):
return f"-ERR {msg}\r\n".encode()
def encode_bulk_string(s):
if s is None:
return b"$-1\r\n"
encoded = s.encode() if isinstance(s, str) else s
return f"${len(encoded)}\r\n".encode() + encoded + b"\r\n"
async def read_resp(reader):
line = await reader.readline()
if not line:
return None
prefix = chr(line[0])
data = line[1:].strip().decode()
if prefix == "+":
return data
elif prefix == "-":
return Exception(data)
elif prefix == ":":
return int(data)
elif prefix == "$":
length = int(data)
if length == -1:
return None
content = await reader.readexactly(length + 2)
return content[:length].decode()
elif prefix == "*":
count = int(data)
if count == -1:
return None
elements = []
for _ in range(count):
element = await read_resp(reader)
elements.append(element)
return elements
else:
return data
def handle_command(command):
if not command or not isinstance(command, list):
return encode_error("invalid command")
cmd = command[0].upper()
args = command[1:]
if cmd == "PING":
return encode_simple_string("PONG")
elif cmd == "ECHO":
if len(args) != 1:
return encode_error("wrong number of arguments for 'echo' command")
return encode_bulk_string(args[0])
else:
return encode_error(f"unknown command '{cmd}'")
async def handle_client(reader, writer):
addr = writer.get_extra_info("peername")
print(f"Client connected: {addr}")
try:
while True:
command = await read_resp(reader)
if command is None:
break
response = handle_command(command)
writer.write(response)
await writer.drain()
except (ConnectionResetError, asyncio.IncompleteReadError):
pass
finally:
print(f"Client disconnected: {addr}")
writer.close()
await writer.wait_closed()
async def main():
server = await asyncio.start_server(handle_client, "localhost", 6379)
addr = server.sockets[0].getsockname()
print(f"Redis server listening on {addr[0]}:{addr[1]}")
async with server:
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())And here is the complete test_client.py:
import socket
import time
def send_resp_command(sock, *args):
cmd = f"*{len(args)}\r\n"
for arg in args:
arg_str = str(arg)
cmd += f"${len(arg_str)}\r\n{arg_str}\r\n"
sock.sendall(cmd.encode())
return sock.recv(4096).decode()
def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("localhost", 6379))
print("--- Test PING ---")
print(repr(send_resp_command(sock, "PING")))
print("\n--- Test ECHO ---")
print(repr(send_resp_command(sock, "ECHO", "hello world")))
sock.close()
if __name__ == "__main__":
main()Your server now speaks the real Redis protocol. Any Redis client that connects on port 6379 can send PING or ECHO and get back valid RESP responses. Next up, you'll add the commands that make Redis actually useful: SET and GET with key expiry.
Implement SET, GET, and Key Expiry
Your server can parse commands and respond to PING and ECHO. Those are useful for testing, but they don't store anything.
A database needs to remember data. In this step, you'll implement SET and GET to turn your server into a real key-value store. You'll also add time-based expiry so keys can automatically disappear after a deadline.
In this step, get ready to:
- Implement the SET command with optional EX and PX expiry arguments.
- Implement the GET command with automatic expiry checking.
- Test the full flow including key expiry with a time delay.
Add the SET command handler
The SET command stores a key-value pair in memory. It also accepts optional EX (seconds) or PX (milliseconds) arguments that set an expiry timestamp on the key.
- In server.py, find the else: block at the bottom of your handle_command function.
- Add the following SET handler directly above that else: block:
elif cmd == "SET":
if len(args) < 2:
return encode_error("wrong number of arguments for 'set' command")
key = args[0]
value = args[1]
store[key] = value
# Clear any existing expiry for this key
if key in expiry:
del expiry[key]
# Parse optional EX/PX expiry arguments
i = 2
while i < len(args):
option = args[i].upper()
if option == "EX" and i + 1 < len(args):
seconds = int(args[i + 1])
expiry[key] = time.time() * 1000 + seconds * 1000
i += 2
elif option == "PX" and i + 1 < len(args):
milliseconds = int(args[i + 1])
expiry[key] = time.time() * 1000 + milliseconds
i += 2
else:
i += 1
return encode_simple_string("OK")What does this code do?
The SET handler stores the key and value in the global store dictionary. If the key already had an expiry, it clears it first so a new SET without EX/PX removes the old deadline.
The while loop walks through any extra arguments after the value. If it finds EX, it converts seconds to a millisecond timestamp. If it finds PX, it uses the millisecond value directly. Both store a future timestamp in the expiry dictionary.
Why store timestamps in milliseconds?
Python's time.time() returns seconds as a float. Multiplying by 1000 converts to milliseconds, which lets both EX and PX share the same comparison logic. This matches how real Redis stores expiry internally.
- Save server.py.
You won't see a change yet. The next chunk adds GET so you can retrieve what you stored.
Add key expiry checking and the GET command handler
Before GET can return a value, it needs to check whether the key has expired. You'll write a small helper function that handles this check, then wire it into the GET handler.
- In server.py, find the blank line between the end of the read_resp function and the start of the handle_command function.
- Add the following is_expired helper function in that gap:
def is_expired(key):
if key in expiry:
if time.time() * 1000 > expiry[key]:
del store[key]
del expiry[key]
return True
return FalseWhat does this code do?
This is passive expiration. Instead of a background thread sweeping for expired keys, you check on access. If the current time (in milliseconds) exceeds the stored deadline, the key is deleted from both store and expiry and the function returns True.
Real Redis uses this same approach for most expired keys. It only adds a background sampling sweep on top for keys that are never accessed again.
- Now find the SET handler you just added inside handle_command.
- Add the following GET handler directly below the SET handler's return encode_simple_string("OK") line, and above the final else: block:
elif cmd == "GET":
if len(args) != 1:
return encode_error("wrong number of arguments for 'get' command")
key = args[0]
# Check if the key has expired before returning
if is_expired(key):
return encode_bulk_string(None)
value = store.get(key)
if value is None:
return encode_bulk_string(None)
return encode_bulk_string(value)What does this code do?
GET first calls is_expired(key) to clean up dead keys before attempting a lookup. If the key was expired (or never existed), it returns the RESP null bulk string $-1\r\n. Otherwise it returns the value as a bulk string.
- Save server.py.
Seeing indentation errors?
Both the SET and GET handlers must be indented at the same level as the existing if cmd == "PING": block inside handle_command. They are elif branches of that same if/elif/else chain.
Make sure is_expired is a standalone function at the module level (no indentation), NOT nested inside another function.
help me fix indentation in my handle_command function
✔️ Awesome, I've got everything!
Great. Double check you've saved server.py before moving on.
ⓧ I'd like to double check the full code
import asyncio
import time
store = {}
expiry = {}
def encode_simple_string(s):
return f"+{s}\r\n".encode()
def encode_error(msg):
return f"-ERR {msg}\r\n".encode()
def encode_bulk_string(s):
if s is None:
return b"$-1\r\n"
encoded = s.encode() if isinstance(s, str) else s
return f"${len(encoded)}\r\n".encode() + encoded + b"\r\n"
async def read_resp(reader):
line = await reader.readline()
if not line:
return None
prefix = chr(line[0])
data = line[1:].strip().decode()
if prefix == "+":
return data
elif prefix == "-":
return Exception(data)
elif prefix == ":":
return int(data)
elif prefix == "$":
length = int(data)
if length == -1:
return None
content = await reader.readexactly(length + 2)
return content[:length].decode()
elif prefix == "*":
count = int(data)
if count == -1:
return None
elements = []
for _ in range(count):
element = await read_resp(reader)
elements.append(element)
return elements
else:
return data
def is_expired(key):
if key in expiry:
if time.time() * 1000 > expiry[key]:
del store[key]
del expiry[key]
return True
return False
def handle_command(command):
if not command or not isinstance(command, list):
return encode_error("invalid command")
cmd = command[0].upper()
args = command[1:]
if cmd == "PING":
return encode_simple_string("PONG")
elif cmd == "ECHO":
if len(args) != 1:
return encode_error("wrong number of arguments for 'echo' command")
return encode_bulk_string(args[0])
elif cmd == "SET":
if len(args) < 2:
return encode_error("wrong number of arguments for 'set' command")
key = args[0]
value = args[1]
store[key] = value
if key in expiry:
del expiry[key]
i = 2
while i < len(args):
option = args[i].upper()
if option == "EX" and i + 1 < len(args):
seconds = int(args[i + 1])
expiry[key] = time.time() * 1000 + seconds * 1000
i += 2
elif option == "PX" and i + 1 < len(args):
milliseconds = int(args[i + 1])
expiry[key] = time.time() * 1000 + milliseconds
i += 2
else:
i += 1
return encode_simple_string("OK")
elif cmd == "GET":
if len(args) != 1:
return encode_error("wrong number of arguments for 'get' command")
key = args[0]
if is_expired(key):
return encode_bulk_string(None)
value = store.get(key)
if value is None:
return encode_bulk_string(None)
return encode_bulk_string(value)
else:
return encode_error(f"unknown command '{cmd}'")
async def handle_client(reader, writer):
addr = writer.get_extra_info("peername")
print(f"Client connected: {addr}")
try:
while True:
command = await read_resp(reader)
if command is None:
break
response = handle_command(command)
writer.write(response)
await writer.drain()
except (ConnectionResetError, asyncio.IncompleteReadError):
pass
finally:
print(f"Client disconnected: {addr}")
writer.close()
await writer.wait_closed()
async def main():
server = await asyncio.start_server(handle_client, "localhost", 6379)
addr = server.sockets[0].getsockname()
print(f"Redis server listening on {addr[0]}:{addr[1]}")
async with server:
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())Update the test client and run the full suite
Your server now supports SET and GET with expiry. Time to prove it works end-to-end.
- Replace the entire main() function in test_client.py with the following updated test suite:
def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("localhost", 6379))
print("--- Test PING ---")
print(repr(send_resp_command(sock, "PING")))
print("\n--- Test ECHO ---")
print(repr(send_resp_command(sock, "ECHO", "hello world")))
print("\n--- Test SET and GET ---")
print(repr(send_resp_command(sock, "SET", "name", "redis")))
print(repr(send_resp_command(sock, "GET", "name")))
print("\n--- Test GET nonexistent key ---")
print(repr(send_resp_command(sock, "GET", "missing")))
print("\n--- Test SET with EX (2 second expiry) ---")
print(repr(send_resp_command(sock, "SET", "temp", "value", "EX", "2")))
print("GET immediately:", repr(send_resp_command(sock, "GET", "temp")))
print("Waiting 3 seconds...")
time.sleep(3)
print("GET after expiry:", repr(send_resp_command(sock, "GET", "temp")))
print("\n--- Test unknown command ---")
print(repr(send_resp_command(sock, "FAKECMD")))
sock.close()
print("\nAll tests complete!")What does this test do?
The test suite exercises each path: SET a key, GET it back, GET a key that doesn't exist (should return null), SET with a 2-second EX expiry, confirm the value is available immediately, wait 3 seconds, then confirm the key has disappeared.
The unknown command test at the end verifies your else branch returns an error response instead of crashing.
- Save test_client.py.
- Start your server by running this in one terminal:
python server.py- Run the test client in a second terminal:
python test_client.py- You should see +OK\r\n for SET, the stored value for GET, $-1\r\n for the missing key, the value returned immediately after the EX set, then $-1\r\n after the 3-second wait.
GET returns null immediately after SET?
Make sure your SET handler stores the value with store[key] = value before parsing EX/PX options. If the store assignment is inside the while loop, it never runs for a plain SET without options.
Also check that the GET handler uses store.get(key) (which returns None for missing keys) rather than store[key] (which raises a KeyError).
help me debug why GET returns null right after SET
✔️ Awesome, I've got everything!
Great. Double check you've saved test_client.py before moving on.
ⓧ I'd like to double check the full code
import socket
import time
def send_resp_command(sock, *args):
cmd = f"*{len(args)}\r\n"
for arg in args:
arg_str = str(arg)
cmd += f"${len(arg_str)}\r\n{arg_str}\r\n"
sock.sendall(cmd.encode())
return sock.recv(4096).decode()
def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(("localhost", 6379))
print("--- Test PING ---")
print(repr(send_resp_command(sock, "PING")))
print("\n--- Test ECHO ---")
print(repr(send_resp_command(sock, "ECHO", "hello world")))
print("\n--- Test SET and GET ---")
print(repr(send_resp_command(sock, "SET", "name", "redis")))
print(repr(send_resp_command(sock, "GET", "name")))
print("\n--- Test GET nonexistent key ---")
print(repr(send_resp_command(sock, "GET", "missing")))
print("\n--- Test SET with EX (2 second expiry) ---")
print(repr(send_resp_command(sock, "SET", "temp", "value", "EX", "2")))
print("GET immediately:", repr(send_resp_command(sock, "GET", "temp")))
print("Waiting 3 seconds...")
time.sleep(3)
print("GET after expiry:", repr(send_resp_command(sock, "GET", "temp")))
print("\n--- Test unknown command ---")
print(repr(send_resp_command(sock, "FAKECMD")))
sock.close()
print("\nAll tests complete!")
if __name__ == "__main__":
main()Your server is now a real key-value store. Keys persist, expire on schedule, and multiple clients can share the same data. Next up, you'll extend it with even more commands in the Secret Mission.
Secret mission
Implement EXISTS, DEL, and KEYS
Your Redis server handles the basics. But real Redis clients expect more than PING, ECHO, SET, and GET. Can you add EXISTS, DEL, and glob-pattern KEYS to make your server even more capable?
Clean Up Your Resources
Clean Up Your Resources
This project runs entirely locally using only the Python standard library, so there are no ongoing costs. Decide whether to keep your project files or delete them entirely.
Resources you used:
- The redis-server project folder on your Desktop (contains server.py and test_client.py).
Keep everything running
No action needed. Choose this if you plan to keep extending your Redis server with more commands or features.
- Your redis-server folder stays on your Desktop, ready for you to pick back up.
- The server only runs when you manually start it with python server.py, so nothing is consuming resources in the background.
Pause - I'll come back to this later
Since the server only runs when you start it manually, there is nothing to pause. Your files sit quietly on disk until you return.
- Make sure the server is stopped. If it is still running in a terminal, press Ctrl+C to stop it.
- Close the project in VS Code if you want to free up screen space.
Delete - I don't want to use this again
Remove the project folder entirely to start fresh.
- Make sure the server is stopped. If it is still running in a terminal, press Ctrl+C to stop it.
- Delete the project folder by running the command for your operating system.
macOS
rm -rf ~/Desktop/redis-serverWindows
rmdir /s /q %USERPROFILE%\Desktop\redis-serverNice Work!
Nice Work!
Nice work! You just built a working Redis server from scratch in Python that speaks the real RESP2 protocol.
You've learned how to:
- Build a TCP server from scratch that binds to port 6379 and handles concurrent connections using Python's asyncio event loop.
- Implement a complete RESP2 protocol parser that encodes and decodes arrays, bulk strings, simple strings, integers, and null values.
- Create an in-memory key-value store with time-based key expiry using EX and PX options.
- Secret Mission: Added integer encoding, array encoding, and glob pattern matching for the KEYS command.
Ready to quiz yourself?