migrate category endpoints to flask-smorest, switching to class based views, blueprints, and marshmallow schemas

Author: piyush-jaiswal

app/__init__.py

@@ -8,18 +8,32 @@
 from dotenv import load_dotenv
 from flasgger import Swagger
 from sqlalchemy import MetaData
+from flask_smorest import Api
+
+
+def register_blueprints():
+    from app.migrated_routes.category import bp as category_bp
+    api.register_blueprint(category_bp, url_prefix="/categories")
 
 
 app = Flask(__name__)
 
 load_dotenv()
+
+# sqlalchemy
 app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv("SQLALCHEMY_DATABASE_URI")
 app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
 
+# jwt
 app.config["JWT_SECRET_KEY"] = os.getenv("JWT_SECRET_KEY")
 app.config["JWT_ACCESS_TOKEN_EXPIRES"] = timedelta(hours=3)
 app.config["JWT_REFRESH_TOKEN_EXPIRES"] = timedelta(days=3)
 
+# flask-smorest
+app.config["API_TITLE"] = "Ecommerce REST API"
+app.config["API_VERSION"] = "v1"
+app.config["OPENAPI_VERSION"] = "3.0.2"
+
 # PostgreSQL-compatible naming convention (to follow the naming convention already used in the DB)
 # https://stackoverflow.com/questions/4107915/postgresql-default-constraint-names
 naming_convention = {
@@ -33,6 +47,9 @@
 db = SQLAlchemy(app, metadata=metadata)
 migrate = Migrate(app, db)
 jwt = JWTManager(app)
+api = Api(app)
+
+register_blueprints()
 
 
 @jwt.expired_token_loader

app/migrated_routes/__init__.py

@@ -0,0 +1,292 @@
+from flask.views import MethodView
+from flask_jwt_extended import jwt_required
+from flask_smorest import Blueprint, abort
+from sqlalchemy import exists
+
+from app import db
+from app.models import (
+    Category,
+    Product,
+    Subcategory,
+    category_subcategory,
+    subcategory_product,
+)
+from app.schemas import (
+    CategoriesOut,
+    CategoryIn,
+    CategoryOut,
+    PaginationArgs,
+    ProductsOut,
+    SubcategoriesOut,
+)
+
+bp = Blueprint("category", __name__)
+
+
+@bp.route("")
+class CategoryCollection(MethodView):
+    init_every_request = False
+
+    @bp.response(200, CategoriesOut)
+    def get(self):
+        """
+        Get All Categories
+        ---
+        tags:
+            - Category
+        description: Get all categories.
+        responses:
+            200:
+                description: A list of categories.
+        """
+        return {"categories": Category.query.all()}
+
+    @jwt_required()
+    @bp.arguments(CategoryIn)
+    @bp.response(201, CategoryOut)
+    def post(self, data):
+        """
+        Create Category
+        ---
+        tags:
+            - Category
+        description: Create a new category.
+        security:
+            - access_token: []
+        requestBody:
+            required: true
+            description: name - Name of the category <br> subcategories - Array of subcategory ids (optional)
+            content:
+                application/json:
+                    schema:
+                        type: object
+                        required:
+                            - name
+                        properties:
+                            name:
+                                type: string
+                            subcategories:
+                                type: array
+                                items:
+                                    type: integer
+        responses:
+            201:
+                description: Category created successfully.
+            400:
+                description: Invalid input.
+            401:
+                description: Token expired, missing or invalid.
+            500:
+                description: Error occurred.
+        """
+        category = Category(name=data["name"])
+
+        if sc_ids := data.get("subcategories"):
+            subcategories = Subcategory.query.filter(Subcategory.id.in_(sc_ids)).all()
+            if len(subcategories) != len(sc_ids):
+                abort(422, message="One or more subcategories not present")
+
+            category.subcategories = subcategories
+
+        db.session.add(category)
+        db.session.commit()
+
+        return category
+
+
+@bp.route("/<int:id>")
+class CategoryById(MethodView):
+    init_every_request = False
+
+    def _get(self, id):
+        return Category.query.get_or_404(id)
+
+    @bp.response(200, CategoryOut)
+    def get(self, id):
+        """
+        Get Category
+        ---
+        tags:
+            - Category
+        description: Get a category by ID.
+        parameters:
+            - in: path
+              name: id
+              required: true
+              type: integer
+              description: Category ID
+        responses:
+            200:
+                description: Category retrieved successfully.
+            404:
+                description: Category not found.
+        """
+        return self._get(id)
+
+    @jwt_required()
+    @bp.arguments(CategoryIn(partial=("name",)))
+    @bp.response(200, CategoryOut)
+    def put(self, data, id):
+        """
+        Update Category
+        ---
+        tags:
+            - Category
+        description: Update an existing category.
+        security:
+            - access_token: []
+        parameters:
+            - in: path
+              name: id
+              required: true
+              type: integer
+              description: Category ID
+        requestBody:
+            required: true
+            description: name - Name of the category (optional) <br> subcategories - Array of subcategory ids
+            content:
+                application/json:
+                    schema:
+                        type: object
+                        properties:
+                            name:
+                                type: string
+                            subcategories:
+                                type: array
+                                items:
+                                    type: integer
+        responses:
+            201:
+                description: Category updated successfully.
+            400:
+                description: Invalid input.
+            404:
+                description: Category not found.
+            500:
+                description: Error occurred.
+        """
+        category = self._get(id)
+        if name := data.get("name"):
+            category.name = name
+
+        if sc_ids := data.get("subcategories"):
+            subcategories = Subcategory.query.filter(Subcategory.id.in_(sc_ids)).all()
+            if len(subcategories) != len(sc_ids):
+                abort(422, message="One or more subcategories not present")
+
+            category.subcategories.extend(subcategories)
+
+        db.session.commit()
+        return category
+
+    @jwt_required()
+    @bp.response(204)
+    def delete(self, id):
+        """
+        Delete Category
+        ---
+        tags:
+            - Category
+        description: Delete a category by ID.
+        security:
+            - access_token: []
+        parameters:
+            - in: path
+              name: id
+              required: true
+              type: integer
+              description: Category ID
+        responses:
+            200:
+                description: Category deleted successfully.
+            404:
+                description: Category not found.
+            500:
+                description: Error occurred.
+        """
+        category = self._get(id)
+        db.session.delete(category)
+        db.session.commit()
+
+
+@bp.route("/<int:id>/subcategories")
+class CategorySubcategories(MethodView):
+    init_every_request = False
+
+    @bp.response(200, SubcategoriesOut)
+    def get(self, id):
+        """
+        Get Subcategories within a Category.
+        ---
+        tags:
+            - Category
+        description: Get Subcategories within a Category.
+        parameters:
+            - in: path
+              name: id
+              required: true
+              type: integer
+              description: Category ID
+        responses:
+            200:
+                description: Subcategories retrieved successfully.
+            404:
+                description: Category not found.
+            500:
+                description: Error occurred.
+        """
+        category = Category.query.get_or_404(id)
+        return {"subcategories": category.subcategories}
+
+
+@bp.route("/<int:id>/products")
+class CategoryProducts(MethodView):
+    init_every_request = False
+    _PER_PAGE = 10
+
+    @bp.arguments(PaginationArgs, location="query", as_kwargs=True)
+    @bp.response(200, ProductsOut)
+    def get(self, id, page):
+        """
+        Get Products within a Category.
+        ---
+        tags:
+            - Category
+        description: Get Products for a Category.
+        parameters:
+            - in: path
+              name: id
+              required: true
+              type: integer
+              description: Category ID
+            - in: query
+              name: page
+              type: integer
+              default: 1
+              description: Page number
+        responses:
+            200:
+                description: Products retrieved successfully.
+            404:
+                description: Category not found.
+            500:
+                description: Error occurred.
+        """
+        category_exists = db.session.query(exists().where(Category.id == id)).scalar()
+        if not category_exists:
+            abort(404)
+
+        products = (
+            Product.query.join(subcategory_product)
+            .join(
+                category_subcategory,
+                onclause=subcategory_product.c.subcategory_id
+                == category_subcategory.c.subcategory_id,
+            )
+            .filter(category_subcategory.c.category_id == id)
+            .distinct()
+            .order_by(Product.id.asc())
+            .paginate(page=page, per_page=CategoryProducts._PER_PAGE, error_out=False)
+        )
+
+        return {"products": products}