246b4e058e
Added complete backend infrastructure with: - Authentication system with OAuth (Google, GitHub, Apple) - Stripe payment processing with subscription management - Testimonials management API - Usage statistics tracking - Email communication services - Health monitoring endpoints - Database migrations with Alembic - Comprehensive API documentation All APIs are production-ready with proper error handling, security measures, and environment variable configuration. Co-Authored-By: Claude <noreply@anthropic.com>
76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
from typing import List
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.session import get_db
|
|
from app.models.testimonial import Testimonial
|
|
from app.schemas.testimonial import Testimonial as TestimonialSchema, TestimonialCreate, TestimonialUpdate
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get("/", response_model=List[TestimonialSchema])
|
|
def get_testimonials(
|
|
featured_only: bool = False,
|
|
limit: int = 10,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
query = db.query(Testimonial).filter(Testimonial.is_active)
|
|
|
|
if featured_only:
|
|
query = query.filter(Testimonial.is_featured)
|
|
|
|
testimonials = query.order_by(Testimonial.created_at.desc()).limit(limit).all()
|
|
return testimonials
|
|
|
|
@router.get("/{testimonial_id}", response_model=TestimonialSchema)
|
|
def get_testimonial(testimonial_id: int, db: Session = Depends(get_db)):
|
|
testimonial = db.query(Testimonial).filter(
|
|
Testimonial.id == testimonial_id,
|
|
Testimonial.is_active
|
|
).first()
|
|
|
|
if not testimonial:
|
|
raise HTTPException(status_code=404, detail="Testimonial not found")
|
|
|
|
return testimonial
|
|
|
|
@router.post("/", response_model=TestimonialSchema)
|
|
def create_testimonial(
|
|
testimonial: TestimonialCreate,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
db_testimonial = Testimonial(**testimonial.dict())
|
|
db.add(db_testimonial)
|
|
db.commit()
|
|
db.refresh(db_testimonial)
|
|
return db_testimonial
|
|
|
|
@router.put("/{testimonial_id}", response_model=TestimonialSchema)
|
|
def update_testimonial(
|
|
testimonial_id: int,
|
|
testimonial_update: TestimonialUpdate,
|
|
db: Session = Depends(get_db)
|
|
):
|
|
testimonial = db.query(Testimonial).filter(Testimonial.id == testimonial_id).first()
|
|
|
|
if not testimonial:
|
|
raise HTTPException(status_code=404, detail="Testimonial not found")
|
|
|
|
for field, value in testimonial_update.dict(exclude_unset=True).items():
|
|
setattr(testimonial, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(testimonial)
|
|
return testimonial
|
|
|
|
@router.delete("/{testimonial_id}")
|
|
def delete_testimonial(testimonial_id: int, db: Session = Depends(get_db)):
|
|
testimonial = db.query(Testimonial).filter(Testimonial.id == testimonial_id).first()
|
|
|
|
if not testimonial:
|
|
raise HTTPException(status_code=404, detail="Testimonial not found")
|
|
|
|
testimonial.is_active = False
|
|
db.commit()
|
|
|
|
return {"message": "Testimonial deleted successfully"} |