Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Simple DB server

Simple DB server

Spec

Write a program that runs a server that is accessible on http://localhost:4000/. When your server receives a request on http://localhost:4000/set?somekey=somevalue it should store the passed key and value in memory. When it receives a request on http://localhost:4000/get?key=somekey it should return the value stored at somekey. During your interview, you will pair on saving the data to a file.

Design

HTTP code conventions chosen:

  • 200 for all successful operations
  • zero-length string and 404 for keys that are not present
  • 500 for anything not in spec

Requirements

See requirements.txt and dev-requirements.txt. Install them using:

pip install -r requirements.txt

Usage

To run the server you must have python and pip installed, then run:

./dbserver.py

To run tests use

pip install -r dev-requirements.txt
python tests.py

Note the persistence test is expected to fail, since restarting the DB server loses DATA.

#!/usr/bin/env python
"""
Simple DB server (key value store) that works over HTTP on localhost.
- GET /set?<key>=<value> Sets <key> to <value>
- GET /get?key=<key> Gets the value for key <key>
Usage:
dbserver.py [--port=<int>]
Options:
-h --help Show this screen.
-p --port=<int> Port to listen on [default: 4000].
"""
from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
from docopt import docopt
import urlparse
DATA = {} # this servers as the in-memory data store
class DBRequestHandlerBase(BaseHTTPRequestHandler):
"""
Handles DB requests using storage-subclass `get` and `set` methods.
"""
def do_GET(self):
"""Base response for all GET requests."""
parsed_path = urlparse.urlparse(self.path)
path = parsed_path.path
#
# 1. validate path
if path not in ['/get', '/set']:
self.send_error(500, 'Error: use /set?<key>=<val> or /get?key=<key>')
return
#
# 2. validate query string
query_dict = urlparse.parse_qs(parsed_path.query)
if len(query_dict) != 1:
self.send_error(500, 'Error: query must have a single field=value pair')
return
query_item = query_dict.items()[0]
q_field, q_value = query_item[0], query_item[1][0]
#
# 3. process request
if path == '/set':
self.set(q_field, q_value) # does actual write to DB
self.respond(http_code=200, http_body='')
elif path == '/get':
if q_field != 'key':
self.send_error(500, 'Error: use /get?key=<keyname> format')
return
value = self.get(q_value) # does actual read from DB
if value is None:
self.respond(http_code=404, http_body='')
return
self.respond(http_code=200, http_body=value)
def respond(self, http_code=200, http_body=''):
"""Respond with default headers the `http_code` and `http_body` given."""
self.send_response(http_code)
self.send_header('Content-type', 'text/plain')
self.end_headers()
self.wfile.write(http_body)
class InMemroryDBRequestHandler(DBRequestHandlerBase):
"""
Uses global dictionary `DATA` as data store for DB functionality.
"""
def get(self, key):
"""Gets key `key` from DB. Returns None if `key` is not present."""
return DATA.get(key, None)
def set(self, key, value):
"""Sets key `key` to value `value` in DB."""
DATA[key] = value
def run_server(port):
"""
The main run loop for the DB server.
"""
server_opts = ('localhost', port)
server = HTTPServer(server_opts, InMemroryDBRequestHandler)
print 'Started Simple DB server on localhost port', port
try:
server.serve_forever()
except KeyboardInterrupt:
server.server_close()
print 'Stopped Simple DB server'
if __name__ == '__main__':
arguments = docopt(__doc__, version='SimpleDB v0.1')
port_int = int(arguments['--port'])
run_server(port=port_int)
from multiprocessing import Process, Event
import requests
import unittest
from BaseHTTPServer import HTTPServer
from dbserver import InMemroryDBRequestHandler
TEST_SERVER_NAME = 'localhost'
TEST_SERVER_PORT = 4001
BASE_URL = 'http://' + TEST_SERVER_NAME + ':' + str(TEST_SERVER_PORT)
def increment_test_server_port():
"""Workaround for Mac OS X holding on to sockets for too long,
which was causing `socket.error: [Errno 48] Address already in use`."""
global TEST_SERVER_PORT, BASE_URL
TEST_SERVER_PORT += 1
BASE_URL = 'http://' + TEST_SERVER_NAME + ':' + str(TEST_SERVER_PORT)
class TestServerProcess(Process):
"""
A special process to workaround the `serve_forever` shenanigans.
Usage:
p = TestServerProcess()
p.start()
# ... run your tests...
p.shutdown()
"""
def __init__(self):
Process.__init__(self)
self.exit = Event()
self.server = HTTPServer(('',TEST_SERVER_PORT), InMemroryDBRequestHandler)
self.server.allow_reuse_address = 1
def silent_logger(self, format, *args):
return
self.server.RequestHandlerClass.log_message = silent_logger
self.daemon = True
def run(self):
while not self.exit.is_set():
self.server.handle_request()
def shutdown(self):
self.exit.set()
self.server.server_close()
increment_test_server_port()
class BasicTestCase(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.server_process = TestServerProcess()
cls.server_process.start()
@classmethod
def tearDownClass(cls):
cls.server_process.shutdown()
def test_basic_set_get(self):
set_response = requests.get(BASE_URL + '/set?key1=val1')
self.assertEqual(200, set_response.status_code)
#
get_response = requests.get(BASE_URL + '/get?key=key1')
self.assertEqual(200, get_response.status_code)
self.assertEqual('val1', get_response.text)
def test_404_missing(self):
response = requests.get(BASE_URL + '/get?key=somethingdoesntexit')
self.assertEqual(404, response.status_code)
def test_500_onmalformed(self):
response = requests.get(BASE_URL + '/malformed')
self.assertEqual(500, response.status_code)
def test_set_overwrites(self):
set_response = requests.get(BASE_URL + '/set?key1=val1')
self.assertEqual(200, set_response.status_code)
#
set_response = requests.get(BASE_URL + '/set?key1=val2')
self.assertEqual(200, set_response.status_code)
#
get_response = requests.get(BASE_URL + '/get?key=key1')
self.assertEqual(200, get_response.status_code)
self.assertEqual('val2', get_response.text)
class DataPersistenceTest(unittest.TestCase):
def setUp(self):
self.start_test_server()
def tearDown(self):
self.stop_test_server()
def start_test_server(self):
self.server_process = TestServerProcess()
self.server_process.start()
def stop_test_server(self):
self.server_process.shutdown()
def test_set_restart_get(self):
set_response = requests.get(BASE_URL + '/set?key1=val1')
self.assertEqual(200, set_response.status_code)
#
self.stop_test_server()
#
self.start_test_server()
#
get_response = requests.get(BASE_URL + '/get?key=key1')
self.assertEqual(200, get_response.status_code)
self.assertEqual('val1', get_response.text)
if __name__ == '__main__':
unittest.main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment