Skip to content

Instantly share code, notes, and snippets.

@dpapathanasiou
Created December 2, 2020 22:50
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save dpapathanasiou/1ddce9b8ef2ad4b14b54ed389cdf338c to your computer and use it in GitHub Desktop.
Save dpapathanasiou/1ddce9b8ef2ad4b14b54ed389cdf338c to your computer and use it in GitHub Desktop.
A simple example of a key-value (kv) storage and lookup service, accessible over http/rest
#!/usr/bin/env python3
"""
simple_kv_service.py
A simple example of a key-value (kv) storage and lookup service,
accessible over http/rest:
http://[host]:[port]/set?somekey=somevalue => assigns "somevalue" to "somekey"
http://[host]:[port]/get?key=somekey => looks up "somekey" and returns its value, if it exists
"""
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlsplit, parse_qsl
from itertools import starmap
class Database:
def __init__(self):
self.db = {}
def get_value(self, key):
print('looking for %s => %s' % (key, self.db.get(key)))
return self.db.get(key)
def set_value(self, key, value):
print('adding: %s => %s' % (key, value))
self.db[key] = value
class Handler(BaseHTTPRequestHandler):
def _get_values(self, keys):
if len(keys) == 0:
return self.send_error(404)
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
for match in map(self.server.db.get_value, keys):
if match:
self.wfile.write(match.encode())
else:
self.wfile.write(b'[no match found!]')
self.wfile.write(b'\r\n')
return
def _set_values(self, tuples):
"""
Since set_value() takes two inputs, itertools.starmap
is simpler than using plain map here.
"""
list(starmap(self.server.db.set_value, tuples))
return
def do_GET(self):
if self.path.startswith("/set"):
# parse_qsl() produces a list of (k,v) tuples from the query string
self._set_values(parse_qsl(urlsplit(self.path).query))
elif self.path.startswith("/get"):
# we only want the values whose key was literally "key"
self._get_values([x[1] for x in parse_qsl(urlsplit(self.path).query) if x[0] == 'key'])
else:
# bad request
self.send_error(400)
class Server:
def __init__(self, host, port, db):
server = HTTPServer((host, port), Handler)
server.db = db
server.serve_forever()
if __name__ == '__main__':
database = Database()
main = Server('', 4000, database)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment