python3 database_server.py
Or, if you prefer virtual environments:
pyvenv venv
source venv/bin/activate
python database_server.py
Navigate to localhost:4000
python3 database_server.py
Or, if you prefer virtual environments:
pyvenv venv
source venv/bin/activate
python database_server.py
Navigate to localhost:4000
import http.server | |
from urllib.parse import urlparse | |
import json | |
DATA = {} | |
class Database(object): | |
''' A simple database object that can get and set data. | |
''' | |
def __init__(self, data_location): | |
# Initializes with a data location | |
self._data_location = data_location | |
# Loads the data into self.data | |
self.data = self._load_data() | |
def _load_data(self): | |
# Makes a copy of the data so that we don't accidentally | |
# blow away our data or modify the original directly. | |
data = self._data_location.copy() | |
return data | |
def _store_data(self): | |
# Storing data merges the old data and the new data | |
self._data_location.update(self.data) | |
return | |
def get(self, key): | |
''' Takes a key to lookup as an argument. | |
Raises an error if the key isn't found. | |
''' | |
try: | |
return self.data[key] | |
except KeyError as error: | |
raise KeyError( | |
'The key `{key}` is not in the database'.format(key=key) | |
) | |
def set(self, key, value): | |
''' Takes a key and a value and sets the key on the data object. | |
''' | |
try: | |
# if the key exists, this is an update, return the status 200 | |
_ = self.get(key) | |
status = 200 | |
except KeyError as err: | |
# if the key didn't exist, we are creating, return a 201 | |
status = 201 | |
self.data[key] = value | |
# Call store data to save the change on the data store | |
self._store_data() | |
return status | |
class DefaultHandler(http.server.BaseHTTPRequestHandler): | |
@property | |
def request_path(self): | |
# Returns the path portion of the URI without the query | |
return urlparse(self.path).path | |
@property | |
def query_string(self): | |
# Returns only the query portion of the path | |
return urlparse(self.path).query | |
@property | |
def db(self): | |
# Returns a connection to the database | |
return Database(DATA) | |
def do_GET(self): | |
if self.request_path == '/': | |
# The root handler returns an dictionary with all of the data | |
return self.respond(200, self.db.data) | |
elif self.request_path == '/set': | |
# First check for the query `?somekey=somevalue`. | |
# Return an error if it is missing | |
if not self.query_string: | |
body = { | |
'title': 'Missing Query', | |
'detail': 'Please provide a query listing the key '\ | |
'(eg. `/get?somekey=somevalue`).' | |
} | |
return self.respond(status_code=400, body=body) | |
# Split the query on the equals sign and assign key and value | |
key, value = self.query_string.split('=') | |
# Set the data in the database and store the status | |
# (ie. was this a create or update event?) | |
status = self.db.set(key, value) | |
# Return success | |
return self.respond(status, {key: value}) | |
elif self.request_path == '/get': | |
# First check for the query `?key=somekey`. | |
# Return an error if it is missing | |
if not self.query_string: | |
body = { | |
'title': 'Missing Query', | |
'detail': 'Please provide a query listing the key '\ | |
'(eg. `/get?key=somekey`).' | |
} | |
return self.respond(status_code=400, body=body) | |
# Split the query | |
split_query = self.query_string.split('=') | |
# If the query isn't requesting a key, return an error | |
if split_query[0] != 'key': | |
body = { | |
'title': 'Query Error', | |
'detail': 'Query must be in the format `/get?key=somekey`.' | |
} | |
return self.respond(status_code=400, body=body) | |
key = split_query[1] | |
try: | |
# Try to fetch the key from the database | |
value = self.db.get(key) | |
except KeyError as err: | |
# If the key doesn't exist, handle the error | |
# and return an error message | |
body = { | |
'title': 'Key Error', | |
'detail': err.args[0] | |
} | |
return self.respond(status_code=400, body=body) | |
else: | |
# Return the data | |
return self.respond(200, {key: value}) | |
else: | |
# 404 handler | |
body = { | |
'title': 'Resource not found', | |
'detail': 'This URL doesn\'t exist!' | |
} | |
return self.respond(status_code=404, body=body) | |
def respond(self, status_code, body=None): | |
''' Respond takes a status code and an optional body. | |
''' | |
if body != None and isinstance(body, (object)): | |
# Checks for the existence of the body, and if it's a dictionary, | |
# sets the content type to JSON and serialize the object. | |
content_type = 'application/json' | |
response_body = json.dumps(body) | |
else: | |
# The default content type is text/html | |
content_type = 'text/html' | |
response_body = body if body else '' | |
# Send headers and set status codes and write the response | |
self.send_response(status_code) | |
self.send_header('Content-type', content_type) | |
self.end_headers() | |
self.wfile.write(response_body.encode('utf-8')) | |
return | |
server = http.server.HTTPServer(('', 4000), DefaultHandler) | |
server.serve_forever() |