Merge pull request #115 from tecladocode/develop

Author: jslvtr

.gitignore

@@ -7,3 +7,5 @@ __pycache__/
 venv/
 .venv/
 docs/docs/.nota/config.ini
+section-start-code.zip
+section-end-code.zip

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/.dockerignore

@@ -0,0 +1,4 @@
+.venv
+*.pyc
+__pycache__
+data.db

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/.flaskenv

@@ -0,0 +1,2 @@
+FLASK_APP=app
+FLASK_DEBUG=True

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3.10
+EXPOSE 5000
+WORKDIR /app
+COPY ./requirements.txt requirements.txt
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+COPY . .
+CMD ["flask", "run", "--host", "0.0.0.0"]

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/app.py

@@ -0,0 +1,105 @@
+from flask import Flask, jsonify
+from flask_smorest import Api
+from flask_jwt_extended import JWTManager
+
+from db import db
+from blocklist import BLOCKLIST
+
+from resources.user import blp as UserBlueprint
+from resources.item import blp as ItemBlueprint
+from resources.store import blp as StoreBlueprint
+from resources.tag import blp as TagBlueprint
+
+
+def create_app(db_url=None):
+    app = Flask(__name__)
+    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 "sqlite:///data.db"
+    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
+    app.config["PROPAGATE_EXCEPTIONS"] = True
+    db.init_app(app)
+    api = Api(app)
+
+    app.config["JWT_SECRET_KEY"] = "jose"
+    jwt = JWTManager(app)
+
+    # @jwt.additional_claims_loader
+    # def add_claims_to_jwt(identity):
+    #     # TODO: Read from a config file instead of hard-coding
+    #     if identity == 1:
+    #         return {"is_admin": True}
+    #     return {"is_admin": False}
+
+    @jwt.token_in_blocklist_loader
+    def check_if_token_in_blocklist(jwt_header, jwt_payload):
+        return jwt_payload["jti"] in BLOCKLIST
+
+    @jwt.expired_token_loader
+    def expired_token_callback(jwt_header, jwt_payload):
+        return (
+            jsonify({"message": "The token has 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,
+        )
+
+    @jwt.revoked_token_loader
+    def revoked_token_callback(jwt_header, jwt_payload):
+        return (
+            jsonify(
+                {"description": "The token has been revoked.", "error": "token_revoked"}
+            ),
+            401,
+        )
+
+    # JWT configuration ends
+
+    with app.app_context():
+        import models  # noqa: F401
+
+        db.create_all()
+
+    api.register_blueprint(UserBlueprint)
+    api.register_blueprint(ItemBlueprint)
+    api.register_blueprint(StoreBlueprint)
+    api.register_blueprint(TagBlueprint)
+
+    return app

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/blocklist.py

@@ -0,0 +1,9 @@
+"""
+blocklist.py
+
+This file just contains the blocklist of the JWT tokens. It will be imported by
+app and the logout resource so that tokens can be added to the blocklist when the
+user logs out.
+"""
+
+BLOCKLIST = set()

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/db.py

@@ -0,0 +1,3 @@
+from flask_sqlalchemy import SQLAlchemy
+
+db = SQLAlchemy()

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/models/__init__.py

@@ -0,0 +1,5 @@
+from models.user import UserModel
+from models.item import ItemModel
+from models.tag import TagModel
+from models.store import StoreModel
+from models.item_tags import ItemsTags

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/models/item.py

@@ -0,0 +1,16 @@
+from db import db
+
+
+class ItemModel(db.Model):
+    __tablename__ = "items"
+
+    id = db.Column(db.Integer, primary_key=True)
+    name = db.Column(db.String(80), unique=False, nullable=False)
+    price = db.Column(db.Float(precision=2), unique=False, nullable=False)
+
+    store_id = db.Column(
+        db.Integer, db.ForeignKey("stores.id"), unique=False, nullable=False
+    )
+    store = db.relationship("StoreModel", back_populates="items")
+
+    tags = db.relationship("TagModel", back_populates="items", secondary="items_tags")

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/models/item_tags.py

@@ -0,0 +1,9 @@
+from db import db
+
+
+class ItemsTags(db.Model):
+    __tablename__ = "items_tags"
+
+    id = db.Column(db.Integer, primary_key=True)
+    item_id = db.Column(db.Integer, db.ForeignKey("items.id"))
+    tag_id = db.Column(db.Integer, db.ForeignKey("tags.id"))

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/models/store.py

@@ -0,0 +1,11 @@
+from db import db
+
+
+class StoreModel(db.Model):
+    __tablename__ = "stores"
+
+    id = db.Column(db.Integer, primary_key=True)
+    name = db.Column(db.String(80), unique=True, nullable=False)
+
+    tags = db.relationship("TagModel", back_populates="store", lazy="dynamic")
+    items = db.relationship("ItemModel", back_populates="store", lazy="dynamic")

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/models/tag.py

@@ -0,0 +1,12 @@
+from db import db
+
+
+class TagModel(db.Model):
+    __tablename__ = "tags"
+
+    id = db.Column(db.Integer, primary_key=True)
+    name = db.Column(db.String(80), unique=False, nullable=False)
+    store_id = db.Column(db.Integer(), db.ForeignKey("stores.id"), nullable=False)
+
+    store = db.relationship("StoreModel", back_populates="tags")
+    items = db.relationship("ItemModel", back_populates="tags", secondary="items_tags")

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/models/user.py

@@ -0,0 +1,9 @@
+from db import db
+
+
+class UserModel(db.Model):
+    __tablename__ = "users"
+
+    id = db.Column(db.Integer, primary_key=True)
+    username = db.Column(db.String(80), unique=True, nullable=False)
+    password = db.Column(db.String(80), nullable=False)

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/requirements.txt

@@ -0,0 +1,7 @@
+Flask-JWT-Extended
+Flask-Smorest
+Flask-SQLAlchemy
+passlib
+marshmallow
+python-dotenv
+gunicorn

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/resources/__init__.py

@@ -0,0 +1,68 @@
+from flask.views import MethodView
+from flask_smorest import Blueprint, abort
+from flask_jwt_extended import jwt_required, get_jwt
+from sqlalchemy.exc import SQLAlchemyError
+
+from db import db
+from models import ItemModel
+from schemas import ItemSchema, ItemUpdateSchema
+
+blp = Blueprint("Items", "items", description="Operations on items")
+
+
+@blp.route("/item/<string:item_id>")
+class Item(MethodView):
+    @jwt_required()
+    @blp.response(200, ItemSchema)
+    def get(self, item_id):
+        item = ItemModel.query.get_or_404(item_id)
+        return item
+
+    @jwt_required()
+    def delete(self, item_id):
+        jwt = get_jwt()
+        if not jwt.get("is_admin"):
+            abort(401, message="Admin privilege required.")
+
+        item = ItemModel.query.get_or_404(item_id)
+        db.session.delete(item)
+        db.session.commit()
+        return {"message": "Item deleted."}
+
+    @blp.arguments(ItemUpdateSchema)
+    @blp.response(200, ItemSchema)
+    def put(self, item_data, item_id):
+        item = ItemModel.query.get(item_id)
+
+        if item:
+            item.price = item_data["price"]
+            item.name = item_data["name"]
+        else:
+            item = ItemModel(id=item_id, **item_data)
+
+        db.session.add(item)
+        db.session.commit()
+
+        return item
+
+
+@blp.route("/item")
+class ItemList(MethodView):
+    @jwt_required()
+    @blp.response(200, ItemSchema(many=True))
+    def get(self):
+        return ItemModel.query.all()
+
+    @jwt_required(fresh=True)
+    @blp.arguments(ItemSchema)
+    @blp.response(201, ItemSchema)
+    def post(self, item_data):
+        item = ItemModel(**item_data)
+
+        try:
+            db.session.add(item)
+            db.session.commit()
+        except SQLAlchemyError:
+            abort(500, message="An error occurred while inserting the item.")
+
+        return item

docs/docs/08_flask_jwt_extended/12_token_refreshing_flask_jwt_extended/end_video/resources/item.py

@@ -0,0 +1,48 @@
+from flask.views import MethodView
+from flask_smorest import Blueprint, abort
+from sqlalchemy.exc import SQLAlchemyError, IntegrityError
+
+from db import db
+from models import StoreModel
+from schemas import StoreSchema
+
+
+blp = Blueprint("Stores", "stores", description="Operations on stores")
+
+
+@blp.route("/store/<string:store_id>")
+class Store(MethodView):
+    @blp.response(200, StoreSchema)
+    def get(self, store_id):
+        store = StoreModel.query.get_or_404(store_id)
+        return store
+
+    def delete(self, store_id):
+        store = StoreModel.query.get_or_404(store_id)
+        db.session.delete(store)
+        db.session.commit()
+        return {"message": "Store deleted"}, 200
+
+
+@blp.route("/store")
+class StoreList(MethodView):
+    @blp.response(200, StoreSchema(many=True))
+    def get(self):
+        return StoreModel.query.all()
+
+    @blp.arguments(StoreSchema)
+    @blp.response(201, StoreSchema)
+    def post(self, store_data):
+        store = StoreModel(**store_data)
+        try:
+            db.session.add(store)
+            db.session.commit()
+        except IntegrityError:
+            abort(
+                400,
+                message="A store with that name already exists.",
+            )
+        except SQLAlchemyError:
+            abort(500, message="An error occurred creating the store.")
+
+        return store