Database Server

.gitignore

venv/
.pytest_cache
__pycache__
*.csv

README.md

Database Server

Before your interview, 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. You can start with simply appending each write to the file, and work on making it more efficient if you have time.

Source

Setup

Prerequisites: Python 3

$ virtualenv venv 
$ pip install -r requirements.txt

I actually used pyenv for the setup, so the command is pyenv activate venv.

Run the App

$ python app.py

By default, the application is accessible at http://localhost:4000

Run Tests

$ pytest -v

app.py

from flask import Flask
from flask import request
from flask import jsonify
import os 
import csv

app = Flask(__name__)

# A simple dictionary is our in-memory data store. This will only work with a 
# single-threaded application on a local machine.
# A potentially better option could be to use the :memory: file in SQlite: 
# https://flask.palletsprojects.com/en/1.1.x/patterns/sqlite3/
KEY_VALUE_PAIRS = {}

CSV_NAME = 'kv_pairs.csv'
CSV_HEADERS = ['key', 'value']

@app.route("/")
def index():
    return """Welcome to my RC Application!"""


@app.route("/set", methods=["GET", "POST"])
def set_kv():
    """retrieves a key-value pair from the client and stores it in memory."""
    data = request.args.to_dict()
    if data == {}:
        return (
            jsonify(
                message="Please submit a key value pair.", statusCode=400, isError=True
            ),
            400,
        )
    data = {str(k): str(v) for k, v in data.items()}
    try:
        KEY_VALUE_PAIRS.update(data)
        write_to_csv(CSV_NAME, data)
    except IOError:
        return jsonify(message='Application error!', statusCode=500, isError=True)
    return (
        jsonify(message="Submitted! {}".format(data), statusCode=200, isError=False,),
        200,
    )

def create_if_not_exists(csv_name):
    if not os.path.exists(csv_name):
        with open(CSV_NAME, 'w', newline='') as csvfile:
            filewriter = csv.writer(csvfile)
            filewriter.writerow(CSV_HEADERS)

def write_to_csv(csv_name, kv_pair):
    create_if_not_exists(csv_name)
    with open(csv_name, 'a', newline='') as csvfile:
        csv_rows = [ {CSV_HEADERS[0]: k, CSV_HEADERS[1]: v}  for k, v in kv_pair.items()]
        dw = csv.DictWriter(csvfile, CSV_HEADERS, -999)
        for row in csv_rows:
            dw.writerow(row)


@app.route("/get", methods=["GET"])
def get_kv():
    """retrieves a key from the client and looks for an associated value in memory."""
    key = request.args.get("key")
    if key is None:
        return (
            jsonify(
                message="Malformed request. Please submit with a 'key' arg",
                statusCode=400,
                isError=True,
            ),
            400,
        )
    value = KEY_VALUE_PAIRS.get(key)
    if value is not None:
        return jsonify(message=value, statusCode=200, isError=False), 200
    return (
        jsonify(message="{} not found".format(value), statusCode=204, isError=False),
        204,
    )


if __name__ == "__main__":
    app.run(debug=True, port=4000)

requirements.txt

appdirs==1.4.3
appnope==0.1.0
attrs==19.3.0
backcall==0.1.0
black==19.10b0
Click==7.0
decorator==4.4.1
Flask==1.1.1
importlib-metadata==1.3.0
ipdb==0.12.3
ipython==7.11.0
ipython-genutils==0.2.0
itsdangerous==1.1.0
jedi==0.15.2
Jinja2==2.10.3
MarkupSafe==1.1.1
more-itertools==8.0.2
packaging==19.2
parso==0.5.2
pathspec==0.7.0
pexpect==4.7.0
pickleshare==0.7.5
pluggy==0.13.1
prompt-toolkit==3.0.2
ptyprocess==0.6.0
py==1.8.1
Pygments==2.5.2
pyparsing==2.4.6
pytest==5.3.2
regex==2019.12.20
six==1.13.0
toml==0.10.0
traitlets==4.3.3
typed-ast==1.4.0
wcwidth==0.1.7
Werkzeug==0.16.0
zipp==0.6.0

test_app.py

from app import app, KEY_VALUE_PAIRS
import unittest


class TestApp(unittest.TestCase):
    def setUp(self):
        self.app = app.test_client()
        KEY_VALUE_PAIRS["recurse"] = "center"

    def test_main(self):
        rv = self.app.get("/")
        assert rv.status_code == 200

    def test_set(self):
        """test we can successfully store POST and store the k-v pair."""
        url = "/set"
        mock_data = "foo=bar"
        response = self.app.post("{}?{}".format(url, mock_data))
        assert response.status_code == 200
        json_data = response.get_json()
        assert "foo" in json_data["message"]

    def test_set_null_args(self):
        """test that null args will produce an error."""
        url = "/set"
        mock_data = ""
        response = self.app.post("{}?{}".format(url, mock_data))
        assert response.status_code == 400

    def test_get(self):
        """"test we can successfully retrieve a k-v pair."""
        url = "/get"
        mock_data = "key=recurse"
        response = self.app.get("{}?{}".format(url, mock_data))
        assert response.status_code == 200
        json_data = response.get_json()
        assert "center" in json_data["message"]

    def test_get_no_key_arg(self):
        """test that a bad arg in the GET will produce an error."""
        url = "/get"
        mock_data = "bad_arg=foo"
        response = self.app.get("{}?{}".format(url, mock_data))
        assert response.status_code == 400