Created
March 9, 2020 10:29
-
-
Save isaacgr/5ceff8d9576547b401c01c9a30436acc to your computer and use it in GitHub Desktop.
Klein json api authentication class
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
from functools import wraps | |
import json | |
from klein import Klein | |
class JsonApi(object): | |
""" | |
Reusable class for composing a JSON API easily with minimal | |
repeated code. | |
""" | |
def __init__(self, app=None, secret='twisted'): | |
self.app = app if app else Klein() | |
self.secret = secret | |
def toJSON(self, data): | |
""" | |
Serialize data to JSON | |
""" | |
return json.dumps(data) | |
def defaultMiddleware(self, f): | |
""" | |
Middleware to set application/json as default header for | |
all responses. | |
""" | |
@wraps(f) | |
def deco(*args, **kwargs): | |
request = args[0] | |
request.setHeader('Content-Type', 'application/json') | |
return self.toJSON(f(*args, **kwargs)) | |
return deco | |
def authenticate(self, f): | |
""" | |
Middleware api-key authentication | |
""" | |
@wraps(f) | |
def deco(*args, **kwargs): | |
request = args[0] | |
apiKey = request.getHeader('Authorization') | |
if not apiKey or apiKey != self.secret: | |
request.setResponseCode(401) | |
body = { | |
'scope': 'private', | |
'message': 'Sorry, valid credentials required to access content'} | |
return body | |
return f(*args, **kwargs) | |
return deco | |
def route(self, url, *args, **kwargs): | |
""" | |
Extend the route functionality | |
""" | |
def deco(f): | |
restricted = kwargs.pop('restricted', False) | |
if restricted: | |
f = self.authenticate(f) | |
f = self.defaultMiddleware(f) | |
self.app.route(url, *args, **kwargs)(f) | |
return deco | |
def main(): | |
main_app = Klein() | |
json_api = JsonApi(app=main_app, secret='tw15t3d') | |
@main_app.route('/') | |
def home(request): | |
return 'Hello World' | |
@json_api.route('/jsonify', methods=['GET']) | |
def jsonify(request): | |
return {'foo': 'bar'} | |
@json_api.route('/admin', restricted=True) | |
def admin(request): | |
return {'scope': 'private', 'message': 'Access restricted content'} | |
@json_api.route('/admin/v2') | |
@json_api.authenticate | |
def admin_v2(request): | |
return {'scope': 'private', 'message': 'Access restricted content'} | |
main_app.run('localhost', 8080) | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment