Skip to content

Instantly share code, notes, and snippets.

@adregan
Last active August 29, 2015 14:24
Show Gist options
  • Save adregan/32b1246c507d6c81647c to your computer and use it in GitHub Desktop.
Save adregan/32b1246c507d6c81647c to your computer and use it in GitHub Desktop.
This simple little database server uses only standard libraries and runs with python 3.

Getting started

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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment