148 lines
4.9 KiB
Python
148 lines
4.9 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status, Form, File, UploadFile
|
|
from sqlmodel import Session, select
|
|
from sqlalchemy.orm import joinedload
|
|
from typing import List, Optional
|
|
from uuid import UUID
|
|
from datetime import datetime
|
|
from app.core.database import get_session
|
|
from app.models.coupon import Coupon
|
|
from app.models.user import User, UserRole
|
|
from app.api.deps import get_current_user
|
|
from app.services.image_handler import save_image, delete_image
|
|
|
|
router = APIRouter(prefix="/api/coupons", tags=["coupons"])
|
|
|
|
@router.get("", response_model=List[Coupon])
|
|
async def list_coupons(
|
|
*,
|
|
session: Session = Depends(get_session),
|
|
active_only: bool = True
|
|
):
|
|
"""List all coupons."""
|
|
statement = select(Coupon).options(joinedload(Coupon.business))
|
|
if active_only:
|
|
statement = statement.where(Coupon.is_active)
|
|
|
|
coupons = session.exec(statement).all()
|
|
return coupons
|
|
|
|
@router.post("", response_model=Coupon)
|
|
async def create_coupon(
|
|
*,
|
|
session: Session = Depends(get_session),
|
|
title: str = Form(...),
|
|
business_id: Optional[UUID] = Form(None),
|
|
description: Optional[str] = Form(None),
|
|
business_name: Optional[str] = Form(None),
|
|
business_address: Optional[str] = Form(None),
|
|
business_phone: Optional[str] = Form(None),
|
|
discount_percentage: Optional[int] = Form(None),
|
|
discount_amount: Optional[float] = Form(None),
|
|
category: Optional[str] = Form(None),
|
|
valid_from: Optional[datetime] = Form(None),
|
|
valid_until: Optional[datetime] = Form(None),
|
|
is_active: bool = Form(True),
|
|
image: Optional[UploadFile] = File(None),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Create a new coupon (Promoters and Admins only)."""
|
|
if current_user.role not in [UserRole.ADMIN, UserRole.PROMOTER]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only promoters and admins can create coupons"
|
|
)
|
|
|
|
image_url = None
|
|
if image:
|
|
image_url = save_image(image, "coupons")
|
|
|
|
db_coupon = Coupon(
|
|
title=title,
|
|
business_id=business_id,
|
|
description=description,
|
|
business_name=business_name,
|
|
business_address=business_address,
|
|
business_phone=business_phone,
|
|
discount_percentage=discount_percentage,
|
|
discount_amount=discount_amount,
|
|
category=category,
|
|
valid_from=valid_from,
|
|
valid_until=valid_until,
|
|
is_active=is_active,
|
|
image_url=image_url
|
|
)
|
|
session.add(db_coupon)
|
|
session.commit()
|
|
session.refresh(db_coupon)
|
|
return db_coupon
|
|
|
|
@router.patch("/{coupon_id}", response_model=Coupon)
|
|
async def update_coupon(
|
|
*,
|
|
session: Session = Depends(get_session),
|
|
coupon_id: UUID,
|
|
title: Optional[str] = Form(None),
|
|
description: Optional[str] = Form(None),
|
|
discount_percentage: Optional[int] = Form(None),
|
|
valid_until: Optional[datetime] = Form(None),
|
|
is_active: Optional[bool] = Form(None),
|
|
image: Optional[UploadFile] = File(None),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Update a coupon (Promoters and Admins only)."""
|
|
if current_user.role not in [UserRole.ADMIN, UserRole.PROMOTER]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only promoters and admins can update coupons"
|
|
)
|
|
|
|
db_coupon = session.get(Coupon, coupon_id)
|
|
if not db_coupon:
|
|
raise HTTPException(status_code=404, detail="Coupon not found")
|
|
|
|
if title is not None:
|
|
db_coupon.title = title
|
|
if description is not None:
|
|
db_coupon.description = description
|
|
if discount_percentage is not None:
|
|
db_coupon.discount_percentage = discount_percentage
|
|
if valid_until is not None:
|
|
db_coupon.valid_until = valid_until
|
|
if is_active is not None:
|
|
db_coupon.is_active = is_active
|
|
|
|
if image:
|
|
if db_coupon.image_url:
|
|
delete_image(db_coupon.image_url)
|
|
db_coupon.image_url = save_image(image, "coupons")
|
|
|
|
session.add(db_coupon)
|
|
session.commit()
|
|
session.refresh(db_coupon)
|
|
return db_coupon
|
|
|
|
@router.delete("/{coupon_id}")
|
|
async def delete_coupon(
|
|
*,
|
|
session: Session = Depends(get_session),
|
|
coupon_id: UUID,
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Delete a coupon (Promoters and Admins only)."""
|
|
if current_user.role not in [UserRole.ADMIN, UserRole.PROMOTER]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Only promoters and admins can delete coupons"
|
|
)
|
|
|
|
db_coupon = session.get(Coupon, coupon_id)
|
|
if not db_coupon:
|
|
raise HTTPException(status_code=404, detail="Coupon not found")
|
|
|
|
if db_coupon.image_url:
|
|
delete_image(db_coupon.image_url)
|
|
|
|
session.delete(db_coupon)
|
|
session.commit()
|
|
return {"status": "success", "message": "Coupon deleted"}
|