resp-server 0.1.0

A lightweight, embeddable, pure-Python Redis-compatible server for testing

RESP Server - Embeddable Redis-Compatible Server for Python

A lightweight, pure-Python implementation of a Redis-compatible server, designed primarily for local development and unit testing.

[!NOTE] Use Case: This is an embedded server (like SQLite) for your tests. It complements redis-py (the client) by giving you a zero-dependency server to connect to without Docker.

Features

• Zero Dependencies: Pure Python standard library (no external dependencies to run the server).
• Embeddable: Run it inside your pytest suite without Docker or external Redis installation.
• RESP Compatible: Works with any Redis client library (redis-py, node-redis, go-redis, etc.).
• Lite: Supports Strings, Lists, Streams, Pub/Sub, and Expiration.

Installation

pip install resp-server

Quick Start

1. Run as a Standalone Server

# Starts server on port 6379
resp-server --port 6379

2. Embed in Python Tests (Pytest Example)

You can spin up the server programmatically in your tests/fixtures:

import pytest
import threading
from resp_server.core.server import Server
import redis

@pytest.fixture(scope="session")
def redis_server():
    # Start server in a background thread
    server = Server(port=6399)
    t = threading.Thread(target=server.start, daemon=True)
    t.start()
    yield server
    server.stop()

def test_my_app(redis_server):
    # Connect using standard client
    r = redis.Redis(port=6399, decode_responses=True)
    r.set("foo", "bar")
    assert r.get("foo") == "bar"

Architecture

Directory Structure

resp_server/
├── main.py                      # CLI Entry point
├── core/
│   ├── server.py                # Server Class & TCP logic
│   ├── command_execution.py     # Command Router
│   ├── datastore.py             # In-Memory DB
├── protocol/
│   ├── resp.py                  # RESP Parser

Codebase Structure & Module Interaction

graph TD
    %% Main Entry Point
    Main["resp_server/main.py<br/>(CLI Entry)"]
    
    %% Core Modules
    Server["resp_server/core/server.py<br/>(TCP Server Class)"]
    CmdExec["resp_server/core/command_execution.py<br/>(Command Router)"]
    Datastore["resp_server/core/datastore.py<br/>(In-Memory DB)"]
    
    %% Utilities
    Parser["resp_server/parser.py<br/>(RESP Parser)"]

    %% Flow Relationships
    Main -->|1. Parses Args & Init| Server
    Server -->|2. Accepts Connections| CmdExec
    
    CmdExec -->|3. Parses Raw Bytes| Parser
    CmdExec -->|4. Read/Write Data| Datastore

---

Features & Internals

This section explains what each feature is, how it works internally, and how to use it.

1. Redis Serialization Protocol (RESP)

• What it is: The binary-safe protocol Redis uses to communicate.
• How it works: The RESP Parser (resp_server/parser.py) reads bytes from the TCP socket, identifying types by their first byte (+ for strings, $ for bulk strings, * for arrays). It recursively parses nested arrays.
• Usage: Transparent to the user. All clients (redis-py, node-redis, etc.) speak this automatically.

2. Strings & Expiration

• What it is: Basic key-value storage with Time-To-Live (TTL).
• How it works: Values are stored in DATA_STORE wrapper dicts: {'value': ..., 'expiry': timestamp}. The get_data_entry function checks the timestamp on every access (lazy expiry) and deletes the key if expired.
• Usage:

  r.set("mykey", "Hello World", px=5000)  # Set with 5s expiry
  r.get("mykey")  # Returns "Hello World"
  

3. Lists

• What it is: Ordered collections of strings.
• How it works: Stored as Python lists in DATA_STORE. Supports push/pop operations from both ends.
• Usage:

  r.rpush("mylist", "A", "B", "C")
  r.lrange("mylist", 0, -1)  # Returns ['A', 'B', 'C']
  

4. Streams

• What it is: An append-only log data structure.
• How it works: Stored in STREAMS as a list of entries. XADD validates IDs (must be incremental). XREAD supports blocking by using a threading.Condition variable to put the client thread to sleep until new data arrives.
• Usage:

  r.xadd("mystream", {"sensor-id": "1234", "temperature": "19.8"})
  r.xrange("mystream", "-", "+")
  

5. Pub/Sub

• What it is: Real-time messaging where publishers send messages to channels and subscribers receive them.
• How it works:
  • CHANNEL_SUBSCRIBERS maps channel names to a Set of client sockets.
  • When PUBLISH is called, the server iterates through the socket list for that channel and writes the message directly to them.
• Usage:

  # Client A
  p = r.pubsub()
  p.subscribe("mychannel")
  
  # Client B
  r.publish("mychannel", "Hello Subscribers!")
  

---

Production Readiness Assessment

⚠️ Current Status: Embedded / Development Use Only

This project is designed as an embedded server for local development and unit testing, similar to SQLite. It is Not Production Ready for critical business data.

Criteria	Status	Notes
Stability	✅ High	Passes standard redis-py integration tests for supported features.
Concurrency	⚠️ Medium	Uses threads (good for I/O), but limited by Python GIL for CPU-bound tasks.
Persistence	⚠️ Partial	Loads RDB files but does not currently implement background saving (BGSAVE).
Security	❌ None	No password authentication (AUTH) or TLS encryption implemented.
Scalability	❌ Low	Single-threaded event loop (conceptually) via Python threads; not async/await based.

---

Usage Examples

Here are real sample outputs from running the server with redis-py:

Starting the Server

$ python3 -m resp_server.main --port 6399
RDB file not found at ./dump.rdb, starting with empty DATA_STORE.
Server: Starting server on localhost:6399...
Server: Listening for connections...

Basic Operations with redis-py

import redis
import time

r = redis.Redis(port=6399, decode_responses=True)

# Connection test
print(f'PING -> {r.ping()}')
# Output: PING -> True

# String operations
print(f'SET mykey "Hello World" -> {r.set("mykey", "Hello World")}')
# Output: SET mykey "Hello World" -> True

print(f'GET mykey -> {r.get("mykey")}')
# Output: GET mykey -> Hello World

# Expiration
print(f'SET temp "I expire in 2s" PX 2000 -> {r.set("temp", "I expire in 2s", px=2000)}')
# Output: SET temp "I expire in 2s" PX 2000 -> True

print(f'GET temp -> {r.get("temp")}')
# Output: GET temp -> I expire in 2s

time.sleep(2.1)
print(f'GET temp (after 2.1s) -> {r.get("temp")}')
# Output: GET temp (after 2.1s) -> None

# List operations
print(f'RPUSH mylist A B C -> {r.rpush("mylist", "A", "B", "C")}')
# Output: RPUSH mylist A B C -> 3

print(f'LRANGE mylist 0 -1 -> {r.lrange("mylist", 0, -1)}')
# Output: LRANGE mylist 0 -1 -> ['A', 'B', 'C']

---

Testing

Run the test suite to verify functionality:

# Install test dependencies
pip install pytest redis

# Run tests
pytest tests/

Sample Output:

============================= test session starts ==============================
collected 13 items

tests/test_datastore.py ........                                         [ 61%]
tests/test_integration.py ...                                            [ 84%]
tests/test_protocol.py ..                                                [100%]

============================== 13 passed in 2.45s ===============================

---

Publishing to PyPI (Optional)

[!NOTE] This section is only relevant if you want to publish your own fork/version to PyPI. Most users can skip this.

Prerequisites

1. Create accounts on PyPI and TestPyPI
2. Generate API tokens:
   • TestPyPI: https://test.pypi.org/manage/account/token/
   • PyPI: https://pypi.org/manage/account/token/

Setup

Create ~/.pypirc with your tokens:

[distutils]
index-servers =
    pypi
    testpypi

[pypi]
repository = https://upload.pypi.org/legacy/
username = __token__
password = pypi-YOUR_PYPI_TOKEN_HERE

[testpypi]
repository = https://test.pypi.org/legacy/
username = __token__
password = pypi-YOUR_TESTPYPI_TOKEN_HERE

Publishing Steps (Using setuptools + twine)

# 1. Install build tools
pip install build twine

# 2. Build the package
python -m build

# 3. Upload to TestPyPI first
twine upload --repository testpypi dist/*

# 4. Verify installation from TestPyPI
pip install --index-url https://test.pypi.org/simple/ resp-server

# 5. If everything works, upload to PyPI
twine upload dist/*

Alternative: Using flit

# 1. Install flit
pip install flit

# 2. Build and publish to TestPyPI
flit publish --repository testpypi

# 3. Publish to PyPI
flit publish

---

Project Origin

This project was built as part of the CodeCrafters Redis Challenge, extended with additional features for portfolio purposes.

License

MIT License - see LICENSE file for details.