Spaces:
Sleeping
Sleeping
File size: 7,726 Bytes
b2be963 | 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 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session, joinedload
from sqlalchemy import desc, asc, func
from app.schemas.product import ProductListItem
from app.db.base import get_db
from app.models.home import HomeSection
from app.models.product import Product, Category
from app.schemas.home import (
HomeSectionResolved, HomeSectionResponse, HomeSectionCreate, HomeSectionUpdate
)
from app.schemas.api_response import APIResponse
from app.core.security import get_current_user
from app.api.auth import get_current_admin_user
from app.models.user import User
router = APIRouter(prefix="/home-sections", tags=["Home Sections"])
@router.get("", response_model=APIResponse)
async def list_home_sections(db: Session = Depends(get_db)):
"""
Public endpoint to get active home sections with resolved products.
"""
sections = db.query(HomeSection).filter(
HomeSection.is_active == True
).order_by(HomeSection.display_order.asc()).all()
resolved_sections = []
for section in sections:
products = []
if section.section_type == "MANUAL" and section.selected_product_ids:
# Fetch specific products by ID
products = db.query(Product).filter(
Product.id.in_(section.selected_product_ids),
Product.is_active == True,
Product.deleted_at == None
).options(joinedload(Product.images), joinedload(Product.category)).all()
# Maintain the manual order
id_map = {p.id: p for p in products}
products = [id_map[pid] for pid in section.selected_product_ids if pid in id_map]
elif section.section_type == "AUTOMATIC" and section.rule:
rule = section.rule
def get_products(rule_dict):
query = db.query(Product).filter(
Product.is_active == True,
Product.deleted_at == None
).options(joinedload(Product.images), joinedload(Product.category))
# Apply filters from rule
if "category_id" in rule_dict:
cat_id = rule_dict["category_id"]
# Get all subcategory IDs recursively
sub_ids = db.query(Category.id).filter(
(Category.id == cat_id) | (Category.parent_id == cat_id)
).all()
# Flatten the list of tuples
all_cat_ids = [r[0] for r in sub_ids]
query = query.filter(Product.category_id.in_(all_cat_ids))
if "is_featured" in rule_dict:
query = query.filter(Product.is_featured == rule_dict["is_featured"])
# Apply sorting
sort_by = rule_dict.get("sort_by", "created_at")
sort_order = rule_dict.get("sort_order", "desc")
if sort_by == "price":
query = query.order_by(asc(Product.price) if sort_order == "asc" else desc(Product.price))
elif sort_by == "rating":
query = query.order_by(desc(Product.rating))
elif sort_by == "discount":
query = query.filter(Product.compare_price > Product.price)
query = query.order_by(desc((Product.compare_price - Product.price) / Product.compare_price))
else:
query = query.order_by(desc(Product.created_at) if sort_order == "desc" else asc(Product.created_at))
limit = rule_dict.get("limit", 8)
return query.limit(limit).all()
products = get_products(rule)
# Fallback: If no products found for a specific rule (like is_featured), try a broader query
if not products:
fallback_rule = rule.copy()
if "is_featured" in fallback_rule:
del fallback_rule["is_featured"]
fallback_rule["sort_by"] = "rating"
products = get_products(fallback_rule)
elif "category_id" in fallback_rule:
# If category is empty, we don't fallback to other categories, but maybe the user wants to see something?
# For now, let's keep category strict but recursive.
pass
# Add primary image_url to each product for the ProductListItem schema
for p in products:
p.image_url = p.images[0].image_url if p.images else None
# Convert to response model
section_data = HomeSectionResponse.model_validate(section).model_dump()
section_data["products"] = [ProductListItem.model_validate(p) for p in products]
resolved_sections.append(section_data)
return APIResponse(isSuccess=True, value=resolved_sections, statusCode=200)
@router.post("/admin", response_model=APIResponse)
async def create_home_section(
section_in: HomeSectionCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_admin_user)
):
section = HomeSection(**section_in.dict())
db.add(section)
db.commit()
db.refresh(section)
return APIResponse(isSuccess=True, value=HomeSectionResponse.model_validate(section), statusCode=201)
@router.get("/admin", response_model=APIResponse)
async def admin_list_sections(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_admin_user)
):
sections = db.query(HomeSection).order_by(HomeSection.display_order.asc()).all()
return APIResponse(isSuccess=True, value=[HomeSectionResponse.model_validate(s) for s in sections], statusCode=200)
@router.put("/admin/{section_id}", response_model=APIResponse)
async def update_home_section(
section_id: int,
section_in: HomeSectionUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_admin_user)
):
section = db.query(HomeSection).filter(HomeSection.id == section_id).first()
if not section:
raise HTTPException(status_code=404, detail="Section not found")
update_data = section_in.dict(exclude_unset=True)
for field, value in update_data.items():
setattr(section, field, value)
db.commit()
db.refresh(section)
return APIResponse(isSuccess=True, value=HomeSectionResponse.model_validate(section), statusCode=200)
@router.delete("/admin/{section_id}", response_model=APIResponse)
async def delete_home_section(
section_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_admin_user)
):
section = db.query(HomeSection).filter(HomeSection.id == section_id).first()
if not section:
raise HTTPException(status_code=404, detail="Section not found")
db.delete(section)
db.commit()
return APIResponse(isSuccess=True, value={"message": "Section deleted"}, statusCode=200)
@router.post("/admin/reorder", response_model=APIResponse)
async def reorder_sections(
orders: List[dict], # List of {"id": 1, "display_order": 0}
db: Session = Depends(get_db),
current_user: User = Depends(get_current_admin_user)
):
for item in orders:
db.query(HomeSection).filter(HomeSection.id == item["id"]).update(
{"display_order": item["display_order"]}
)
db.commit()
return APIResponse(isSuccess=True, value={"message": "Order updated"}, statusCode=200)
|