Skip to content

Instantly share code, notes, and snippets.

@rscarrera27
Created July 11, 2019 01:33
Show Gist options
  • Save rscarrera27/e75822ac896c548b0de9eb2f9febf6a4 to your computer and use it in GitHub Desktop.
Save rscarrera27/e75822ac896c548b0de9eb2f9febf6a4 to your computer and use it in GitHub Desktop.
from sanic import Sanic
from sanic.response import json
from sanic.request import Request
from sanic_jwt_extended import (
JWTManager, jwt_required, create_access_token,
create_refresh_token, jwt_refresh_token_required)
from sanic_jwt_extended.tokens import Token
from sanic_jwt_extended.blacklists import InMemoryBlacklist
app = Sanic(__name__)
# Setup the Sanic-JWT-Extended extension
app.config['JWT_SECRET_KEY'] = 'super-secret' # Change this!
app.config['JWT_ACCESS_BLACKLIST_ENABLE'] = True
app.config['JWT_REFRESH_BLACKLIST_ENABLE'] = True
refresh_blacklist = InMemoryBlacklist()
access_blacklist = InMemoryBlacklist()
app.config['JWT_ACCESS_BLACKLIST'] = access_blacklist
app.config['JWT_REFRESH_BLACKLIST'] = refresh_blacklist
JWTManager(app)
@app.route('/login', methods=['POST'])
async def login(request: Request):
username = request.json.get('username', None)
password = request.json.get('password', None)
access_token = await create_access_token(identity=username, app=request.app)
refresh_token = await create_refresh_token(identity=username, app=request.app)
return json(
dict(
access_token=access_token,
refresh_token=refresh_token
),
status=200)
@app.route("/refresh", methods=['GET'])
@jwt_refresh_token_required(blacklist_check=True)
async def refresh(request: Request, token: Token):
return json(dict(access_token=await create_access_token(identity=token.jwt_identity, app=request.app)), status=200)
@app.route('/protected', methods=['GET'])
@jwt_required(blacklist_check=True)
async def protected(request: Request, token: Token):
current_user = token.raw_jwt
return json(dict(data=current_user))
@app.route('/refrsh-logout', methods=['GET'])
@jwt_refresh_token_required
async def logout(request: Request, token: Token):
refresh_blacklist.add(token)
# or request.app.jwt.refresh_blacklist.add(token)
return json({"msg": "logout succeed"})
@app.route('/access-logout', methods=['GET'])
@jwt_required
async def logout(request: Request, token: Token):
access_blacklist.add(token)
# or request.app.jwt.refresh_blacklist.add(token)
return json({"msg": "logout succeed"})
if __name__ == '__main__':
app.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment