Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
from http import HTTPStatus
from http.client import HTTPException, HTTPResponse, parse_headers
from http.server import SimpleHTTPRequestHandler
import re
import json
from urllib.request import Request
from urllib.parse import urljoin
from http.cookies import SimpleCookie, BaseCookie
import logging
import io
import sys
rule_re = re.compile(r'/([A-z0-9_]+|\<\w+(:\w+)?\>)')
identifier_re = re.compile(r'\<(?P<name>\w+)(?P<type>:\w+)?\>')
format='%(asctime)s - %(levelname)s - %(message)s',
level=logging.DEBUG, stream=sys.stdout,
logger = logging.getLogger(__name__)
def _mkdct(f):
dct = {
'name': f.__name__,
'doc': f.__doc__ if f.__doc__ else 'No documentation.',
if hasattr(f, '__rule__'):
if hasattr(f, '__allowedmethods__'):
if hasattr(f, '__statuscode__'):
return dct
class HTTPStatusException(HTTPException):
def __init__(self, status_code, message=None):
if message is None:
message = HTTPStatus(status_code).name
self.status_code = status_code
self.message = message
def __str__(self):
return f'{self.status_code} : {self.message}'
class RESTRequestHandler(SimpleHTTPRequestHandler):
rules = {}
errorhandlers = {}
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.body = io.BytesIO()
def request_url(self):
host, port = self.server.server_address
return urljoin(f'http://{host}:{port}/', self.path)
def response(self):
Constructs a response from the _header_buffer and the body attributes.
:return: A response as would be sent by urllib.request.urlopen
:rtype: http.client.HTTPResponse
sock = self.server.socket
method = self.command
debuglevel = 4
response = HTTPResponse(sock, debuglevel, method, self.request_url)
headers = b''.join(self._headers_buffer)
response.fp = io.BytesIO(headers +
response.headers = parse_headers(response.fp)
return response
def make_cookie_header(cls, cookie, *attrs):
if isinstance(cookie, BaseCookie):
# cookies is a list or tuple of cookies
if not attrs:
attrs = None
return cookie.output(header='', attrs=attrs)
raise TypeError('`cookie` should be an http.cookies.BaseCookie instance.')
def make_cookie(name, value, domain=None, expires=None, max_age=None, path=None):
cookie = SimpleCookie()
cookie[name] = value
if domain:
cookie[name]['domain'] = domain
if expires:
cookie[name]['expires'] = expires
if max_age:
cookie[name]['max-age'] = max_age
if path:
cookie[name]['path'] = path
return cookie
def route(cls, rule, *, methods=None):
if methods is None:
methods = ['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'HEAD', 'OPTIONS']
def idrepl(match):
data = match.groupdict()
repls = {
'int': r'\d+',
'float': r'\d+\.?\d*',
'str': r'\w+',
'uuid': r'[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{4}-[a-fA-F0-9]{12}'
repl = repls.get(data['type'][1:], r'.+')
return '(?P<%s>%s)' % (data['name'], repl)
def typecaster(match):
# The match includes the leading ':'
type_ = match.groupdict()['type'][1:]
if type_ is None:
return str
tcs = {
'int': int,
'float': float,
'str': str,
'uuid': str,
return tcs.get(type_, str)
def wrapper(func):
func.__allowedmethods__ = [m.upper() for m in methods]
func.__rule__ = rule
# Construct a regex according to the rule
# That regex will match the type of the rule, and its name will be
# the same as the resulting capturing group.
func.__typecast__ = {match.groupdict()['name']: typecaster(match) for match in identifier_re.finditer(rule)}
regex = re.compile(identifier_re.sub(idrepl, rule))
func.__rulere__ = regex
cls.rules[regex] = func
return func
return wrapper
def errorhandler(cls, status_code=None):
if status_code is not None and status_code != '*':
status_code = HTTPStatus[status_code] if isinstance(status_code, str) else HTTPStatus(status_code)
status_code = '*'
def wrapper(func):
nonlocal status_code
cls.errorhandlers[status_code] = func
func.__statuscode__ = status_code if isinstance(status_code, str) else status_code.value
return func
return wrapper
def handle_requests(self):
callback = None
has_callback = False
for rule, cb in self.rules.items():
if rule.fullmatch(self.path):
if self.command in cb.__allowedmethods__:
callback = cb
has_callback = True
# Handle cases when the rule is undefined, or the method is not allowed.
if callback is None:
if has_callback:
return ''
# Create a Request object to hold all the data from the request
# That object is always the first argument of the callback
request = Request(
if 'Cookie' in self.headers:
request.cookies = {
c.split('=')[0]: c.split('=')[1]
for c in self.headers['Cookie'].split('; ')
# Gets the kwargs for the callback function, based on the identifiers in
# the rule. Typecast them into the correct type subsequently.
kwargs = callback.__rulere__.match(self.path).groupdict()
kwargs = {name: callback.__typecast__.get(name, str)(v) for name, v in kwargs.items()}
rv = callback(request, **kwargs)
except HTTPStatusException as e:
if e.status_code in self.errorhandlers:
rv = self.errorhandlers[HTTPStatus(e.status_code)](request, e)
elif '*' in self.errorhandlers:
rv = self.errorhandlers['*'](request, e)
self.send_error(e.status_code, e.message)
data = ''
headers = {}
cookies = []
if isinstance(rv, (HTTPStatus, int)):
status_code = rv
elif len(rv) == 2:
status_code, data = rv
elif len(rv) == 3:
status_code, data, headers = rv
elif len(rv) == 4:
status_code, data, headers, cookies = rv
raise HTTPException(f'Invalid number of return arguments for {callback.__name__}')
headers.update({'Last-Modified': self.date_time_string()})
# Write the body of the response
self.body = io.BytesIO()
if data:
if isinstance(data, (dict, list)):
headers.update({'Content-Type': 'application/json'})
data = json.dumps(data)
data = data.encode('utf8')
headers.update({'Content-Length': len(data)})
# Send the headers.
for header_name, header_value in headers.items():
self.send_header(header_name, header_value)
for cookie in cookies:
if isinstance(cookie, BaseCookie):
cookie = [cookie]
self.send_header('Set-Cookie', self.make_cookie_header(*cookie))
if self.request_version != 'HTTP/0.9':
if hasattr(self, '_headers_buffer'):
# Send the body of the request
self.copyfile(self.body, self.wfile)
self._headers_buffer = []
do_GET = handle_requests
do_POST = handle_requests
do_PUT = handle_requests
do_PATCH = handle_requests
do_DELETE = handle_requests
do_HEAD = handle_requests
do_OPTIONS = handle_requests
@RESTRequestHandler.route('/doc', methods=['GET'])
def doc(request):
"""Returns information about all available routes"""
endpoints = {
'rules': [_mkdct(func) for func in RESTRequestHandler.rules.values()],
'errorhandlers': [_mkdct(func) for func in RESTRequestHandler.errorhandlers.values()],
return HTTPStatus.OK, endpoints
@RESTRequestHandler.route('/doc/<name:str>', methods=['GET'])
def doc_info(request, name):
"""Returns information about routes with a callback matching `name`"""
endpoints = {
'rules': [_mkdct(func) for func in RESTRequestHandler.rules.values() if name.lower() in func.__name__.lower()],
'errorhandlers': [_mkdct(func) for func in RESTRequestHandler.errorhandlers.values() if name.lower() in func.__name__.lower()],
return HTTPStatus.OK, endpoints
if __name__ == '__main__':
from http.server import HTTPServer
import argparse
__version__ = '0.1.1'
parser = argparse.ArgumentParser(prog='webhookserver')
parser.add_argument('--version', '-V', action='version', version=f'apiserver v{__version__}')
parser.add_argument('--verbose', '-v', action='store_true')
group = parser.add_mutually_exclusive_group()
hpgrp = group.add_argument_group()
hpgrp.add_argument('--host', default='localhost')
hpgrp.add_argument('--port', type=int, default='8080')
group.add_argument('--bind', '-b', default='localhost:8080')
args = parser.parse_args()
if 'bind' in vars(args):
address = args.bind.split(':')
address = address[0], int(address[1])
address =, args.port
if not args.verbose:
logger.handlers = [logging.NullHandler(logging.INFO)]'Serving server on http://%s:%s' % address)
@RESTRequestHandler.route('/get', methods=['GET'])
def get(r):
return HTTPStatus.OK, {'foo': 'bar'}
@RESTRequestHandler.route('/post', methods=['POST'])
def post(r):
print(f'Received POST request : {}')
return HTTPStatus.OK
@RESTRequestHandler.route('/put', methods=['PUT'])
def put(r):
print(f'Received PUT request : {}')
@RESTRequestHandler.route('/delete', methods=['DELETE'])
def delete(r):
print('Received DELETE request')
@RESTRequestHandler.route('/get/<obj_id:int>', methods=['GET'])
def get_obj(r, obj_id):
objs = [
{'bar': 'baz'},
{'lorem': 'ipsum'},
if obj_id > len(objs):
raise HTTPStatusException(HTTPStatus.NOT_FOUND)
return HTTPStatus.OK, objs[obj_id]
def error(r, code):
raise HTTPStatusException(code)
def post_user(r, user_id, post_uuid):
return HTTPStatus.OK, {'uid': user_id, 'post': post_uuid}
def set_cookie(r, name, value):
"""Returns a dict view of currently set cookies, and set a cookie's name and value"""
return HTTPStatus.OK, {'cookies': r.cookies}, {}, [RESTRequestHandler.make_cookie(name, value)]
def handle_errors(r, exc):
return exc.status_code, {'code': exc.status_code, 'message': exc.message}, {'Content-Type': 'application/json'}
def handle_403(r, exc):
return exc.status_code, {'code': exc.status_code, 'message': 'You are not allowed to access this content.'}, {'Content-Type': 'application/json'}
with HTTPServer(address, RESTRequestHandler) as httpd:
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment