from utils import render, Log
from models import Product
import json
class Product_api(object):
def list(self, ctx):
"""GET /api/products - List all products"""
try:
request = ctx.getRequest()
# Parse query parameters
page = int(request.getParameter("page") or "1")
limit = int(request.getParameter("limit") or "10")
category = request.getParameter("category")
search = request.getParameter("search")
# Validate pagination parameters
if page < 1: page = 1
if limit < 1 or limit > 100: limit = 10
# Build query
query = "1=1"
params = []
if category:
query += " AND category = ?"
params.append(category)
if search:
query += " AND (name LIKE ? OR description LIKE ?)"
params.extend(["%%%s%%" % search, "%%%s%%" % search])
# Get total count for pagination
total_count = Product.count(query, *params)
# Calculate pagination
offset = (page - 1) * limit
total_pages = (total_count + limit - 1) // limit
# Fetch products
products = Product.where(query, *params).offset(offset).limit(limit).orderBy("created_at DESC")
# Format response
product_list = []
for product in products:
product_list.append({
"id": product.get("id"),
"name": product.get("name"),
"description": product.get("description"),
"price": float(product.get("price") or 0),
"category": product.get("category"),
"stock_quantity": product.get("stock_quantity"),
"is_active": product.getBoolean("is_active"),
"created_at": str(product.get("created_at")),
"updated_at": str(product.get("updated_at"))
})
response = {
"success": True,
"data": product_list,
"pagination": {
"page": page,
"limit": limit,
"total_count": total_count,
"total_pages": total_pages,
"has_next": page < total_pages,
"has_prev": page > 1
}
}
ctx.go_to = render.as_json(ctx, response)
except Exception as e:
Log.error(ctx, "Error in product list API: %s" % str(e))
self._error_response(ctx, "Internal server error", 500)
def get(self, ctx):
"""GET /api/products/{id} - Get single product"""
try:
request = ctx.getRequest()
product_id = request.getParameter("id")
if not product_id:
return self._error_response(ctx, "Product ID is required", 400)
product = Product.findById(product_id)
if not product:
return self._error_response(ctx, "Product not found", 404)
response = {
"success": True,
"data": {
"id": product.get("id"),
"name": product.get("name"),
"description": product.get("description"),
"price": float(product.get("price") or 0),
"category": product.get("category"),
"stock_quantity": product.get("stock_quantity"),
"is_active": product.getBoolean("is_active"),
"created_at": str(product.get("created_at")),
"updated_at": str(product.get("updated_at"))
}
}
ctx.go_to = render.as_json(ctx, response)
except Exception as e:
Log.error(ctx, "Error in product get API: %s" % str(e))
self._error_response(ctx, "Internal server error", 500)
def create(self, ctx):
"""POST /api/products - Create new product"""
try:
request = ctx.getRequest()
# Extract and validate required fields
name = request.getParameter("name")
price = request.getParameter("price")
category = request.getParameter("category")
if not name or len(name.strip()) < 2:
return self._error_response(ctx, "Product name is required (min 2 characters)", 400)
if not price:
return self._error_response(ctx, "Price is required", 400)
try:
price_float = float(price)
if price_float < 0:
return self._error_response(ctx, "Price must be positive", 400)
except ValueError:
return self._error_response(ctx, "Invalid price format", 400)
if not category:
return self._error_response(ctx, "Category is required", 400)
# Create product
product = Product()
product.set("name", name.strip())
product.set("description", request.getParameter("description") or "")
product.set("price", price_float)
product.set("category", category)
product.set("stock_quantity", int(request.getParameter("stock_quantity") or "0"))
product.set("is_active", request.getParameter("is_active") == "true")
product.save()
response = {
"success": True,
"message": "Product created successfully",
"data": {
"id": product.get("id"),
"name": product.get("name"),
"description": product.get("description"),
"price": float(product.get("price")),
"category": product.get("category"),
"stock_quantity": product.get("stock_quantity"),
"is_active": product.getBoolean("is_active"),
"created_at": str(product.get("created_at"))
}
}
ctx.getResponse().setStatus(201) # Created
ctx.go_to = render.as_json(ctx, response)
except Exception as e:
Log.error(ctx, "Error in product create API: %s" % str(e))
self._error_response(ctx, "Internal server error", 500)
def update(self, ctx):
"""PUT /api/products/{id} - Update product"""
try:
request = ctx.getRequest()
product_id = request.getParameter("id")
if not product_id:
return self._error_response(ctx, "Product ID is required", 400)
product = Product.findById(product_id)
if not product:
return self._error_response(ctx, "Product not found", 404)
# Update fields if provided
name = request.getParameter("name")
if name:
if len(name.strip()) < 2:
return self._error_response(ctx, "Product name must be at least 2 characters", 400)
product.set("name", name.strip())
price = request.getParameter("price")
if price:
try:
price_float = float(price)
if price_float < 0:
return self._error_response(ctx, "Price must be positive", 400)
product.set("price", price_float)
except ValueError:
return self._error_response(ctx, "Invalid price format", 400)
description = request.getParameter("description")
if description is not None:
product.set("description", description)
category = request.getParameter("category")
if category:
product.set("category", category)
stock_quantity = request.getParameter("stock_quantity")
if stock_quantity is not None:
try:
product.set("stock_quantity", int(stock_quantity))
except ValueError:
return self._error_response(ctx, "Invalid stock quantity", 400)
is_active = request.getParameter("is_active")
if is_active is not None:
product.set("is_active", is_active == "true")
product.save()
response = {
"success": True,
"message": "Product updated successfully",
"data": {
"id": product.get("id"),
"name": product.get("name"),
"description": product.get("description"),
"price": float(product.get("price")),
"category": product.get("category"),
"stock_quantity": product.get("stock_quantity"),
"is_active": product.getBoolean("is_active"),
"updated_at": str(product.get("updated_at"))
}
}
ctx.go_to = render.as_json(ctx, response)
except Exception as e:
Log.error(ctx, "Error in product update API: %s" % str(e))
self._error_response(ctx, "Internal server error", 500)
def delete(self, ctx):
"""DELETE /api/products/{id} - Delete product"""
try:
request = ctx.getRequest()
product_id = request.getParameter("id")
if not product_id:
return self._error_response(ctx, "Product ID is required", 400)
product = Product.findById(product_id)
if not product:
return self._error_response(ctx, "Product not found", 404)
product.delete()
response = {
"success": True,
"message": "Product deleted successfully"
}
ctx.go_to = render.as_json(ctx, response)
except Exception as e:
Log.error(ctx, "Error in product delete API: %s" % str(e))
self._error_response(ctx, "Internal server error", 500)
def _error_response(self, ctx, message, status_code=400):
"""Helper method to generate error responses"""
response = {
"success": False,
"error": {
"message": message,
"code": status_code
}
}
ctx.getResponse().setStatus(status_code)
ctx.go_to = render.as_json(ctx, response)