Created
December 2, 2020 22:50
-
-
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
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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