Skip to content

Instantly share code, notes, and snippets.

@dimitrov
Created June 7, 2013 05:51
Show Gist options
  • Save dimitrov/5727288 to your computer and use it in GitHub Desktop.
Save dimitrov/5727288 to your computer and use it in GitHub Desktop.
#!/usr/bin/env python
import signal
import logging
import base64
import routes
import routes.middleware
import webob
import webob.dec
import webob.exc
import jinja2
import gevent.pywsgi
import simplejson as json
from uuid import uuid4
templates = {'jinja': '''<!DOCTYPE html>
<html>
<head><title>Welcome here</title></head>
<body>
<h1>Welcome</h1>
The value is {{ value|default('not set') }}.
</body>
</html>'''}
LOG = logging.getLogger(__name__)
logging.basicConfig(level=logging.DEBUG, format="%(asctime)s %(msg)s")
DEBUG = True
SUPPORTED_CONTENT_TYPES = ('application/json', 'application/xml')
class JSONSerializer(object):
"""Wrapper around simplejson.dumps()"""
def serialize(self, data):
return json.dumps(data)
class JSONDeserializer(object):
"""Wrapper around simplejson.loads()"""
def deserialize(self, data):
return json.joads(data)
class XMLSerializer(object):
def serialize(self, data):
raise NotImplementedError()
class XMLDeserializer(object):
def deserialize(self, data):
raise NotImplementedError()
class RequestDeserializer(object):
"""This class is responsible for the deserialization of request
objects. It determines the appropriate deserializer based on the
request content type.
"""
def __init__(self):
self.deserializers = {
'application/json': JSONDeserializer(),
'application/xml': XMLDeserializer()
}
def deserialize(self, request):
deserializer = self.deserializers.get(request.content_type)
request.body = deserializer.deserialize(request.body)
return request
class ResponseSerializer(object):
"""This class is responsible for the serialization of response
objects. It determines the appropriate serializer based on the
response content type.
"""
def __init__(self):
self.serializers = {
'application/json': JSONSerializer(),
'application/xml': XMLSerializer()
}
def serialize(self, response, data):
serializer = self.serializers.get(response.content_type)
response.body = serializer.serialize(data)
return response
class APIServer(object):
"""Simple API server"""
def __init__(self):
self.mapper = routes.Mapper()
self.mapper.connect('/v1/', resource='index')
self.mapper.connect('/v1/view/{item}', resource='view')
self.mapper.connect('/', resource='test')
self._router = routes.middleware.RoutesMiddleware(self._dispatch,
self.mapper)
self.default_content_type = 'application/json'
self.serializer = ResponseSerializer()
self.jinja_env = jinja2.Environment(loader=jinja2.DictLoader(
templates))
def index(self, request):
body = {'status': 'ok', 'request_id': str(uuid4())}
return self._build_response(request, body)
def test(self, request):
template = self.jinja_env.get_template('jinja')
return webob.Response(template.render({'value': request.GET.get(
'value', 'none')}))
def _build_response(self, request, data):
resp = webob.Response()
resp.content_type = request.content_type or \
self.default_content_type
return self.serializer.serialize(resp, data)
def check_auth(self, username, password):
return username == 'admin' and password == 'password'
def _process_request(self, request, app):
auth_token = request.headers.get('X-Auth-Token')
if not auth_token or auth_token != 'open-sesame':
raise webob.exc.HTTPForbidden()
return app(request)
def _authenticate(self, request, app):
if request.authorization:
auth_type, credentials = request.authorization
if not auth_type.lower() == 'basic':
raise webob.exc.HTTPNotImplemented()
username, password = base64.b64decode(credentials).split(':', 1)
if self.check_auth(username, password):
return self._process_request(request, app)
return webob.Response('Unauthorized!', 401, [('WWW-Authenticate',
'Basic realm="Login Required"')])
@webob.dec.wsgify
def _dispatch(self, request):
if DEBUG:
request.headers['X-Auth-Token'] = 'open-sesame'
print request
try:
resource = request.urlvars.get('resource')
if resource:
resp = self._authenticate(request, getattr(self, resource))
else:
raise webob.exc.HTTPNotFound()
except AttributeError:
raise webob.exc.HTTPNotFound()
return resp
@webob.dec.wsgify
def __call__(self, request):
return self._router
if __name__ == '__main__':
print 'Serving on 8088...'
LOG.info('Server running...')
gevent.signal(signal.SIGQUIT, gevent.shutdown)
try:
gevent.pywsgi.WSGIServer(('', 8088), APIServer()).serve_forever()
except KeyboardInterrupt:
exit(0)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment