Hi all,
I have two services and want to use the same 20% off coupon code for both. I see this is not yet available to want to start a thread to build interest to have this feature added.
Surely this is only a quick dev fix to allow duplicates???
If it helps 😁
# pip install fastapi uvicorn sqlalchemy pydantic
from datetime import datetime
from typing import List, Optional
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, conint
from sqlalchemy import (
create_engine, Column, Integer, String, Boolean, DateTime,
Numeric, ForeignKey, UniqueConstraint
)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
DATABASE_URL = "sqlite:///./app.db" # replace with your DB
engine = create_engine(DATABASE_URL, future=True)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
Base = declarative_base()
class Coupon(Base):
__tablename__ = "coupons"
id = Column(Integer, primary_key=True)
code = Column(String(64), unique=True, index=True, nullable=False)
discount_type = Column(String(16), nullable=False) # "amount" or "percent"
amount = Column(Numeric(10, 2), nullable=True)
percent = Column(Numeric(5, 2), nullable=True)
starts_at = Column(DateTime, nullable=True)
ends_at = Column(DateTime, nullable=True)
max_redemptions = Column(Integer, nullable=True) # total cap across services
one_per_customer = Column(Boolean, default=True)
active = Column(Boolean, default=True)
services = relationship("CouponService", back_populates="coupon", cascade="all, delete-orphan")
class Service(Base):
__tablename__ = "services"
id = Column(Integer, primary_key=True)
name = Column(String(128), nullable=False)
price = Column(Numeric(10, 2), nullable=False)
coupons = relationship("CouponService", back_populates="service", cascade="all, delete-orphan")
class CouponService(Base):
__tablename__ = "coupon_services"
id = Column(Integer, primary_key=True)
coupon_id = Column(Integer, ForeignKey("coupons.id", ondelete="CASCADE"), nullable=False)
service_id = Column(Integer, ForeignKey("services.id", ondelete="CASCADE"), nullable=False)
__table_args__ = (UniqueConstraint("coupon_id", "service_id", name="uq_coupon_service"),)
coupon = relationship("Coupon", back_populates="services")
service = relationship("Service", back_populates="coupons")
class Redemption(Base):
__tablename__ = "redemptions"
id = Column(Integer, primary_key=True)
coupon_id = Column(Integer, ForeignKey("coupons.id", ondelete="CASCADE"), nullable=False)
service_id = Column(Integer, ForeignKey("services.id", ondelete="CASCADE"), nullable=False)
customer_id = Column(String(128), nullable=False)
used_at = Column(DateTime, default=datetime.utcnow, nullable=False)
Base.metadata.create_all(engine)
# Pydantic I/O
class DuplicateCouponIn(BaseModel):
coupon_code: str
service_ids: List[conint(gt=0)]
class AttachCouponIn(BaseModel):
coupon_id: int
service_ids: List[conint(gt=0)]
app = FastAPI()
def attach_coupon_to_services(db, coupon_id: int, service_ids: List[int]) -> int:
# Verify coupon exists
coupon = db.get(Coupon, coupon_id)
if not coupon:
raise HTTPException(status_code=404, detail="Coupon not found")
# Filter to only existing services
existing_services = db.query(Service.id).filter(Service.id.in_(service_ids)).all()
valid_ids = {sid for (sid,) in existing_services}
if not valid_ids:
return 0
# Find already attached service ids
already = db.query(CouponService.service_id).filter(
CouponService.coupon_id == coupon_id,
CouponService.service_id.in_(valid_ids)
).all()
already_set = {sid for (sid,) in already}
to_add = [sid for sid in valid_ids if sid not in already_set]
for sid in to_add:
db.add(CouponService(coupon_id=coupon_id, service_id=sid))
return len(to_add)
@app.post("/coupons/duplicate-by-code")
def duplicate_coupon_by_code(payload: DuplicateCouponIn):
with SessionLocal() as db:
coupon = db.query(Coupon).filter(Coupon.code == payload.coupon_code).first()
if not coupon:
raise HTTPException(status_code=404, detail="Coupon code not found")
added = attach_coupon_to_services(db, coupon.id, payload.service_ids)
db.commit()
return {"coupon_id": coupon.id, "code": coupon.code, "attached_new_services": added}
@app.post("/coupons/attach")
def attach_coupon(payload: AttachCouponIn):
with SessionLocal() as db:
added = attach_coupon_to_services(db, payload.coupon_id, payload.service_ids)
db.commit()
return {"coupon_id": payload.coupon_id, "attached_new_services": added}
# Validation helper for checkout
def validate_coupon_for_service(db, code: str, service_id: int, customer_id: Optional[str]) -> Coupon:
now = datetime.utcnow()
coupon = db.query(Coupon).filter(Coupon.code == code, Coupon.active == True).first()
if not coupon:
raise HTTPException(status_code=400, detail="Invalid coupon")
if coupon.starts_at and now < coupon.starts_at:
raise HTTPException(status_code=400, detail="Coupon not started")
if coupon.ends_at and now > coupon.ends_at:
raise HTTPException(status_code=400, detail="Coupon expired")
# Check linkage to service
linked = db.query(CouponService).filter(
CouponService.coupon_id == coupon.id,
CouponService.service_id == service_id
).first()
if not linked:
raise HTTPException(status_code=400, detail="Coupon not valid for this service")
# Enforce caps
if coupon.max_redemptions is not None:
total_used = db.query(Redemption).filter(Redemption.coupon_id == coupon.id).count()
if total_used >= coupon.max_redemptions:
raise HTTPException(status_code=400, detail="Coupon redemption limit reached")
if coupon.one_per_customer and customer_id:
used_by_customer = db.query(Redemption).filter(
Redemption.coupon_id == coupon.id,
Redemption.service_id == service_id,
Redemption.customer_id == customer_id
).count()
if used_by_customer > 0:
raise HTTPException(status_code=400, detail="Coupon already used by this customer")
return coupon
# Example apply flow
class ApplyCouponIn(BaseModel):
code: str
service_id: int
customer_id: Optional[str] = None
@app.post("/checkout/apply-coupon")
def apply_coupon(payload: ApplyCouponIn):
with SessionLocal() as db:
coupon = validate_coupon_for_service(db, payload.code, payload.service_id, payload.customer_id)
# Record redemption when you actually capture payment, not here
return {"coupon_id": coupon.id, "code": coupon.code, "valid": True}
