-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathapp.py
132 lines (108 loc) · 3.87 KB
/
app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
"""_summary_
File contains API requests for stores
"""
import os
import secrets
from flask import Flask, jsonify
from flask_smorest import Api
from flask_migrate import Migrate
from db import db
from blocklist import BLOCKLIST
import models
from resources.item import blp as ItemBlueprint
from resources.store import blp as StoreBlueprint
from resources.tag import blp as TagBlueprint
from resources.user import blp as UserBlueprint
from flask_jwt_extended import JWTManager
def create_app(db_url:str=None) -> Flask:
"""_summary_
Args:
db_url (str, optional): The URL of the database to use. Defaults to None.
Returns:
Flask: A Flask application
"""
app = Flask(__name__)
app.config["PROPAGATE_EXCEPTIONS"] = True
app.config["API_TITLE"] = "Stores REST API"
app.config["API_VERSION"] = "v1"
app.config["OPENAPI_VERSION"] = "3.0.3"
app.config["OPENAPI_URL_PREFIX"] = "/"
app.config["OPENAPI_SWAGGER_UI_PATH"] = "/swagger-ui"
app.config["OPENAPI_SWAGGER_UI_URL"] = "https://cdn.jsdelivr.net/npm/swagger-ui-dist/"
app.config["SQLALCHEMY_DATABASE_URI"] = db_url or os.getenv("DATABASE_URL", "sqlite:///data.db")
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db.init_app(app)
migrate = Migrate(app, db)
api = Api(app)
app.config["JWT_SECRET_KEY"] = "236520528094713753437932268324630142015"
jwt = JWTManager(app)
@jwt.token_in_blocklist_loader
def check_if_token_in_blocklist(jwt_header, jwt_payload):
""" Runs whenever we receive a JWT and checks if the token is in the blocklist,
if it is, then the request is terminated and it is revoked
Args:
jwt_header (_type_): _description_
jwt_payload (_type_): _description_
Returns:
_type_: _description_
"""
return jwt_payload["jti"] in BLOCKLIST
@jwt.revoked_token_loader
def revoked_token_callback(jwt_header, jwt_payload):
"""Event handler that returns an error to the user in the event that
check_if_token_in_blocklist returns true
Args:
jwt_header (_type_): _description_
jwt_payload (_type_): _description_
"""
return(
jsonify(
{"description": "The token has been revoked.", "error": "token_revoked"}
), 401
)
# Add a claim to a JWT
@jwt.additional_claims_loader
def add_claims_to_jwt(identity):
# Purely for learning purposes, normally would check in database for a user's permissions
if identity == 1:
return { "is_admin": True }
return { "is_admin": False }
@jwt.expired_token_loader
def expired_token_callback(jwt_header, jwt_payload):
return (
jsonify({
"message": "The token is expired.",
"error": "token_expired"
}), 401
)
@jwt.invalid_token_loader
def invalid_token_callback(error):
return (
jsonify({
"message": "Signature verification failed.",
"error": "invalid_token"
}), 401
)
@jwt.unauthorized_loader
def missing_token_callback(error):
return (
jsonify({
"description": "Request does not contain an access token.",
"error": "authorization_required"
}), 401
)
@jwt.needs_fresh_token_loader
def token_not_fresh_callback(jwt_header, jwt_payload):
return (
jsonify(
{
"description": "The token is not fresh.",
"error": "fresh_token_required"
}
), 401
)
api.register_blueprint(ItemBlueprint)
api.register_blueprint(StoreBlueprint)
api.register_blueprint(TagBlueprint)
api.register_blueprint(UserBlueprint)
return app