File size: 5,053 Bytes
80feb1b |
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 |
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from datetime import datetime, timezone
from ..database import get_db, LoyaltyProgram as LoyaltyProgramModel
from ..models.loyalty import LoyaltyProgram, LoyaltyProgramCreate, LoyaltyProgramUpdate
router = APIRouter(
prefix="/loyalty",
tags=["loyalty"],
responses={404: {"description": "Not found"}},
)
# Get all loyalty program tiers
@router.get("/", response_model=List[LoyaltyProgram])
def get_all_loyalty_tiers(db: Session = Depends(get_db)):
return db.query(LoyaltyProgramModel).order_by(LoyaltyProgramModel.visit_count).all()
# Get active loyalty program tiers
@router.get("/active", response_model=List[LoyaltyProgram])
def get_active_loyalty_tiers(db: Session = Depends(get_db)):
return (
db.query(LoyaltyProgramModel)
.filter(LoyaltyProgramModel.is_active == True)
.order_by(LoyaltyProgramModel.visit_count)
.all()
)
# Get loyalty tier by ID
@router.get("/{tier_id}", response_model=LoyaltyProgram)
def get_loyalty_tier(tier_id: int, db: Session = Depends(get_db)):
db_tier = (
db.query(LoyaltyProgramModel).filter(LoyaltyProgramModel.id == tier_id).first()
)
if not db_tier:
raise HTTPException(status_code=404, detail="Loyalty tier not found")
return db_tier
# Create new loyalty tier
@router.post("/", response_model=LoyaltyProgram)
def create_loyalty_tier(tier: LoyaltyProgramCreate, db: Session = Depends(get_db)):
# Check if a tier with this visit count already exists
existing_tier = (
db.query(LoyaltyProgramModel)
.filter(LoyaltyProgramModel.visit_count == tier.visit_count)
.first()
)
if existing_tier:
raise HTTPException(
status_code=400,
detail=f"Loyalty tier with visit count {tier.visit_count} already exists",
)
# Create new tier
db_tier = LoyaltyProgramModel(
visit_count=tier.visit_count,
discount_percentage=tier.discount_percentage,
is_active=tier.is_active,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc),
)
db.add(db_tier)
db.commit()
db.refresh(db_tier)
return db_tier
# Update loyalty tier
@router.put("/{tier_id}", response_model=LoyaltyProgram)
def update_loyalty_tier(
tier_id: int, tier_update: LoyaltyProgramUpdate, db: Session = Depends(get_db)
):
db_tier = (
db.query(LoyaltyProgramModel).filter(LoyaltyProgramModel.id == tier_id).first()
)
if not db_tier:
raise HTTPException(status_code=404, detail="Loyalty tier not found")
# Check if updating visit count and if it already exists
if (
tier_update.visit_count is not None
and tier_update.visit_count != db_tier.visit_count
):
existing_tier = (
db.query(LoyaltyProgramModel)
.filter(
LoyaltyProgramModel.visit_count == tier_update.visit_count,
LoyaltyProgramModel.id != tier_id,
)
.first()
)
if existing_tier:
raise HTTPException(
status_code=400,
detail=f"Loyalty tier with visit count {tier_update.visit_count} already exists",
)
db_tier.visit_count = tier_update.visit_count
# Update other fields if provided
if tier_update.discount_percentage is not None:
db_tier.discount_percentage = tier_update.discount_percentage
if tier_update.is_active is not None:
db_tier.is_active = tier_update.is_active
db_tier.updated_at = datetime.now(timezone.utc)
db.commit()
db.refresh(db_tier)
return db_tier
# Delete loyalty tier
@router.delete("/{tier_id}")
def delete_loyalty_tier(tier_id: int, db: Session = Depends(get_db)):
db_tier = (
db.query(LoyaltyProgramModel).filter(LoyaltyProgramModel.id == tier_id).first()
)
if not db_tier:
raise HTTPException(status_code=404, detail="Loyalty tier not found")
db.delete(db_tier)
db.commit()
return {"message": "Loyalty tier deleted successfully"}
# Get applicable discount for a visit count
@router.get("/discount/{visit_count}")
def get_discount_for_visit_count(visit_count: int, db: Session = Depends(get_db)):
# Find the tier that exactly matches the visit count
applicable_tier = (
db.query(LoyaltyProgramModel)
.filter(
LoyaltyProgramModel.visit_count == visit_count,
LoyaltyProgramModel.is_active == True,
)
.first()
)
if not applicable_tier:
return {"discount_percentage": 0, "message": "No applicable loyalty discount"}
return {
"discount_percentage": applicable_tier.discount_percentage,
"tier_id": applicable_tier.id,
"visit_count": applicable_tier.visit_count,
"message": f"Loyalty discount of {applicable_tier.discount_percentage}% applied for {visit_count} visits",
}
|