This is a chunk of code I wrote for a job interview at some point - it is a remote key-value storage REST API supporting CRUD operations.
Maybe anyone has some feedback on this? It would be very helpful for me.
I did not attach the tests, since it might be a little bit too much code. You don't have to review both files - even commenting on just one can be extremely helpful.
remotekvs.py
# -*- test-case-name: tests.remotekvs_test.RemoteKVSTest -*-
import json
import inspect
from twisted.internet import reactor
from twisted.web.resource import Resource
from twisted.web.server import Site
from keyvaluestorage import KeyValueStorage
class RemoteKVSResource(Resource):
"""
This is a Twisted Resource object used to serve requests for
KeyValueStorage operations. It supports POST, GET, PUT and DELETE
request methods and replies in JSON.
"""
isLeaf = True
def __init__(self):
self.kvs = KeyValueStorage()
def render(self, request):
operations = {
'POST': 'create',
'GET': 'read',
'PUT': 'update',
'DELETE': 'delete'}
if request.method in operations.keys():
response = {
'success': True}
try:
operation = getattr(self.kvs, operations[request.method])
args = {k: v[0] if isinstance(v, list) else v
for k, v in request.args.items()}
result = operation(**(args))
if result is not None:
response['result'] = result
except (TypeError, KeyValueStorage.KeyAlreadyExistsError,
KeyValueStorage.InvalidKeyError), exception:
response['success'] = False
response['error'] = exception.__class__.__name__
if isinstance(exception, TypeError):
response['error_message'] = (
"Valid request parameters: %s." % (
', '.join(inspect.getargspec(operation)[0][1:])))
else:
response['error_message'] = str(exception)
else:
response = {
'success': False,
'error': 'InvalidRequestMethod',
'error_message': (
"Valid request methods: POST, GET, PUT, DELETE.")}
return json.dumps(response)
if __name__ == '__main__': # pragma: no cover
site = Site(RemoteKVSResource())
reactor.listenTCP(2048, site)
reactor.run()
keyvaluestorage.py
# -*- test-case-name: tests.keyvaluestorage_test.KeyValueStorageTest -*-
class KeyValueStorage(object):
"""The most basic key-value storage, supports all 4 CRUD operations."""
class KeyAlreadyExistsError(Exception):
"""Key you are trying to create already exists in the database."""
class InvalidKeyError(Exception):
"""Key you are trying to access does not exist in the database."""
def __init__(self):
self.storage = {}
def create(self, key, value):
if key in self.storage.keys():
raise self.KeyAlreadyExistsError(
("Key already exists. KeyValueStorage does not allow "
"duplicate keys."))
self.storage[key] = value
def read(self, key):
try:
return self.storage[key]
except KeyError:
raise self.InvalidKeyError(
"Key does not exist. You can not read by a nonexistent key.")
def update(self, key, value):
if key not in self.storage.keys():
raise self.InvalidKeyError(
"Key does not exist. You can not update by a nonexistent key.")
self.storage[key] = value
def delete(self, key):
try:
del self.storage[key]
except KeyError:
raise self.InvalidKeyError(
"Key does not exist. You can not delete by a nonexistent key.")