Skip to content

Feature/flask smorest product #23

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Oct 3, 2025
Merged

Feature/flask smorest product #23

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8e9ab0a
migrate product endpoints to flask-smorest
piyush-jaiswal Oct 3, 2025
10aa701
update product tests and add duplicate duplicate links test
piyush-jaiswal Oct 3, 2025
c1acd3e
update README with new product URLs
piyush-jaiswal Oct 3, 2025
33caec4
fix bug of stripping null strings
piyush-jaiswal Oct 3, 2025
f20532f
allow setting description to null or empty
piyush-jaiswal Oct 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ Test the API using Swagger UI (`/` route), Postman, cURL or your preferred HTTP
### Endpoints

#### Fetch products using name, category, subcategory
- [GET] `/product/<name: string>` - Get product with name: `name` <br/><br/>
- [GET] `/products?name=<name: string>` - Get product with name: `name` <br/><br/>
- [GET] `/subcategories/<subcategory_id: int>/products?page=<page_no>` - Get product with within subcategory `subcategory`. Returns `page_no` of the paginated results. <br/><br/>
- [GET] `/categories/<category_id: int>/products` - Get product with within category `category`. Returns first page of the paginated results. <br/><br/>
- [GET] `/categories/<category_id: int>/products?page=<page_no>` - Get product with within category `category`. Returns `page_no` of the paginated results. <br/><br/>
Expand Down Expand Up @@ -151,11 +151,11 @@ Test the API using Swagger UI (`/` route), Postman, cURL or your preferred HTTP

#### Product
- [GET] `/products` - Get all products
- [GET] `/product/(int: product_id)` - Get product with product_id
- [GET] `/product/(int: product_id)/subcategories` - Get subcategories related to product_id
- [DELETE] `/product/(int: product_id)` (Protected) - Delete product with product_id
- [GET] `/products/(int: product_id)` - Get product with product_id
- [GET] `/products/(int: product_id)/subcategories` - Get subcategories related to product_id
- [DELETE] `/products/(int: product_id)` (Protected) - Delete product with product_id

- [POST] `/product/create` (Protected) - Create a new product
- [POST] `/products` (Protected) - Create a new product
```
{
"name": "name",
Expand All @@ -164,7 +164,7 @@ Test the API using Swagger UI (`/` route), Postman, cURL or your preferred HTTP
}
```

- [PUT] `/product/(int: product_id)/update` (Protected) - Update product with product_id
- [PUT] `/products/(int: product_id)` (Protected) - Update product with product_id
```
{
"name": "name",
Expand Down
2 changes: 2 additions & 0 deletions app/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,11 @@
def register_blueprints():
from app.migrated_routes.category import bp as category_bp
from app.migrated_routes.subcategory import bp as subcategory_bp
from app.migrated_routes.product import bp as product_bp

api.register_blueprint(category_bp, url_prefix="/categories")
api.register_blueprint(subcategory_bp, url_prefix="/subcategories")
api.register_blueprint(product_bp, url_prefix="/products")


app = Flask(__name__)
Expand Down
311 changes: 311 additions & 0 deletions app/migrated_routes/product.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,311 @@
from flask.views import MethodView
from flask_jwt_extended import jwt_required
from flask_smorest import Blueprint, abort
from psycopg2.errors import UniqueViolation
from sqlalchemy import UniqueConstraint
from sqlalchemy.exc import IntegrityError

from app import db
from app.models import (
Product,
Subcategory,
subcategory_product,
)
from app.schemas import (
NameArgs,
PaginationArgs,
ProductIn,
ProductOut,
ProductsOut,
SubcategoriesOut,
)

bp = Blueprint("product", __name__)


@bp.route("")
class ProductCollection(MethodView):
init_every_request = False
_PER_PAGE = 10

@staticmethod
def _get_name_unique_constraint():
name_col = Product.__table__.c.name
return next(
con
for con in Product.__table__.constraints
if isinstance(con, UniqueConstraint)
and len(con.columns) == 1
and con.columns.contains_column(name_col)
)

_NAME_UNIQUE_CONSTRAINT = _get_name_unique_constraint()

def _get_by_name(self, name):
return Product.query.filter(Product.name == name)

@bp.arguments(NameArgs, location="query", as_kwargs=True)
@bp.arguments(PaginationArgs, location="query", as_kwargs=True)
@bp.response(200, ProductsOut)
def get(self, name, page):
"""
Get All Products
---
tags:
- Product
description: Get all products.
parameters:
- in: query
name: page
type: integer
default: 1
description: Page number
- in: query
name: name
type: string
description: Name
responses:
200:
description: Product by name or a paginated list of all products.
"""
if name is not None:
products = self._get_by_name(name)
else:
products = Product.query.order_by(Product.id.asc()).paginate(
page=page, per_page=ProductCollection._PER_PAGE, error_out=False
)

return {"products": products}

@jwt_required()
@bp.arguments(ProductIn)
@bp.response(201, ProductOut)
def post(self, data):
"""
Create Product
---
tags:
- Product
description: Create a new product.
security:
- access_token: []
requestBody:
required: true
description: name - Name of the product <br> description - Description of the product (optional) <br> subcategories - Array of subcategory ids (optional)
content:
application/json:
schema:
type: object
required:
- name
properties:
name:
type: string
description:
type: string
subcategories:
type: array
items:
type: integer
responses:
201:
description: Product created successfully.
400:
description: Invalid input.
401:
description: Token expired, missing or invalid.
500:
description: Error occurred.
"""
product = Product(name=data["name"], description=data.get("description"))

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")
product.subcategories = subcategories

try:
db.session.add(product)
db.session.commit()
except IntegrityError as ie:
db.session.rollback()
if (
isinstance(ie.orig, UniqueViolation)
and ie.orig.diag.constraint_name
== ProductCollection._NAME_UNIQUE_CONSTRAINT.name
):
abort(409, message="Product with this name already exists")
raise
Comment on lines +128 to +139
Copy link
Contributor

@coderabbitai coderabbitai bot Oct 3, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Return 409 on SQLite unique violations instead of leaking 500s

Both post and put handlers only translate unique violations when ie.orig is a psycopg2 UniqueViolation. Under the in-memory SQLite DB used in tests (and commonly in local runs), ie.orig is a sqlite3.IntegrityError, so these blocks fall through to raise. That bubbles the DB error, yielding a 500 instead of the intended 409 and forcing the tests to pytest.raises(IntegrityError). Please broaden the handling to cover SQLite by inspecting ie.orig (e.g., isinstance(..., sqlite3.IntegrityError) plus message matching) before aborting with the appropriate 409 message for duplicate product names or duplicate product-subcategory links.

Also applies to: 236-250

🤖 Prompt for AI Agents 
In app/migrated_routes/product.py around lines 128-139 (and similarly at
236-250), the except block only treats psycopg2 UniqueViolation as a duplicate
and re-raises other DB errors (causing 500s under SQLite); update the exception
handling to also detect SQLite unique constraint violations by checking if
ie.orig is an instance of sqlite3.IntegrityError and/or by matching its message
(e.g., "UNIQUE constraint failed" or the specific constraint/column names) and,
when it indicates the Product name or product-subcategory unique constraint,
call abort(409, message="Product with this name already exists") or the
appropriate 409 message; keep the existing rollback, preserve behavior for other
errors by re-raising, and apply the same change to the put/post handling at
lines 236-250.


return product


@bp.route("/<int:id>")
class ProductById(MethodView):
init_every_request = False

def _get(self, id):
return Product.query.get_or_404(id)

@bp.response(200, ProductOut)
def get(self, id):
"""
Get Product
---
tags:
- Product
description: Get a product by ID.
parameters:
- in: path
name: id
required: true
type: integer
description: Product ID
responses:
200:
description: Product retrieved successfully.
404:
description: Product not found.
"""
return self._get(id)

@jwt_required()
@bp.arguments(ProductIn(partial=("name",)))
@bp.response(200, ProductOut)
def put(self, data, id):
"""
Update Product
---
tags:
- Product
description: Update an existing product.
security:
- access_token: []
consumes:
- application/json
parameters:
- in: path
name: id
required: true
type: integer
description: Product ID
requestBody:
required: true
description: name - Name of the product (optional) <br> description = Description of the product (optional) <br> subcategories - Array of subcategory ids (optional)
content:
application/json:
schema:
type: object
properties:
name:
type: string
description:
type: string
subcategories:
type: array
items:
type: integer
responses:
201:
description: Product updated successfully.
400:
description: Invalid input.
404:
description: Product not found.
500:
description: Error occurred.
"""
product = self._get(id)

if name := data.get("name"):
product.name = name
if "description" in data:
product.description = data["description"]

with db.session.no_autoflush:
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")
product.subcategories.extend(subcategories)

try:
db.session.commit()
except IntegrityError as ie:
db.session.rollback()
if (
isinstance(ie.orig, UniqueViolation)
and ie.orig.diag.constraint_name
== ProductCollection._NAME_UNIQUE_CONSTRAINT.name
):
abort(409, message="Product with this name already exists")
if (
isinstance(ie.orig, UniqueViolation)
and ie.orig.diag.constraint_name == subcategory_product.primary_key.name
):
abort(409, message="Product and subcategory already linked")
raise

return product

@jwt_required()
@bp.response(204)
def delete(self, id):
"""
Delete Product
---
tags:
- Product
description: Delete a product by ID.
security:
- access_token: []
parameters:
- in: path
name: id
required: true
type: integer
description: Product ID
responses:
200:
description: Product deleted successfully.
404:
description: Product not found.
500:
description: Error occurred.
"""
product = self._get(id)
db.session.delete(product)
db.session.commit()


@bp.route("/<int:id>/subcategories")
class ProductSubcategories(MethodView):
init_every_request = False

@bp.response(200, SubcategoriesOut)
def get(self, id):
"""
Get Subcategories related to a Product.
---
tags:
- Product
description: Get Subcategories related to a Product.
parameters:
- in: path
name: id
required: true
type: integer
description: Product ID
responses:
200:
description: Subcategories retrieved successfully.
404:
description: Product not found.
500:
description: Error occurred.
"""
product = Product.query.get_or_404(id)
return {"subcategories": product.subcategories}
Loading