Created
July 11, 2019 01:33
-
-
Save rscarrera27/e75822ac896c548b0de9eb2f9febf6a4 to your computer and use it in GitHub Desktop.
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 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