Coverage for app / api / routes / campaigns.py: 95%

1from typing import Optional, List 

2from fastapi import APIRouter, Depends, HTTPException, Query 

3from sqlalchemy.ext.asyncio import AsyncSession 

4from sqlalchemy import select, func, or_ 

5from sqlalchemy.orm import selectinload 

6 

7from app.db.database import get_db 

8from app.db.models import Campaign, Creator, CampaignCategory, CampaignStatus, Donation, Advocacy 

9from app.schemas.campaign import CampaignCreate, CampaignUpdate, CampaignResponse, CampaignListResponse, CampaignDetailResponse 

10from app.schemas.donation import DonationResponse, DonationListResponse 

11from app.api.deps import get_required_creator 

12 

13router = APIRouter() 

14 

15 

16def calculate_advocate_count(campaign: Campaign) -> int: 

17 """Calculate active advocate count for a campaign.""" 

18 return len([a for a in campaign.advocacies if a.is_active]) 

19 

20 

21@router.get("", response_model=CampaignListResponse) 

22async def list_campaigns( 

23 page: int = Query(1, ge=1), 

24 per_page: int = Query(20, ge=1, le=100), 

25 category: Optional[CampaignCategory] = None, 

26 search: Optional[str] = None, 

27 sort: str = Query("newest", pattern="^(newest|most_advocates|trending)$"), 

28 db: AsyncSession = Depends(get_db), 

29): 

30 """List campaigns with filtering and pagination.""" 

31 query = select(Campaign).where(Campaign.status == CampaignStatus.ACTIVE) 

32 

33 if category: 

34 query = query.where(Campaign.category == category) 

35 

36 if search: 

37 search_term = f"%{search}%" 

38 query = query.where( 

39 or_( 

40 Campaign.title.ilike(search_term), 

41 Campaign.description.ilike(search_term), 

42 ) 

43 ) 

44 

45 # Get total count 

46 count_query = select(func.count()).select_from(query.subquery()) 

47 total_result = await db.execute(count_query) 

48 total = total_result.scalar() or 0 

49 

50 # Apply sorting 

51 if sort == "newest": 

52 query = query.order_by(Campaign.created_at.desc()) 

53 elif sort == "most_advocates": 

54 # Count active advocacies and sort by that count 

55 query = query.outerjoin(Advocacy, (Advocacy.campaign_id == Campaign.id) & (Advocacy.is_active == True)) 

56 query = query.group_by(Campaign.id).order_by(func.count(Advocacy.id).desc(), Campaign.created_at.desc()) 

57 elif sort == "trending": 

58 # Trending = recent campaigns with advocates (sort by advocate count, then recency) 

59 query = query.outerjoin(Advocacy, (Advocacy.campaign_id == Campaign.id) & (Advocacy.is_active == True)) 

60 query = query.group_by(Campaign.id).order_by(func.count(Advocacy.id).desc(), Campaign.created_at.desc()) 

61 

62 # Apply pagination 

63 query = query.offset((page - 1) * per_page).limit(per_page) 

64 

65 # Load relationships 

66 query = query.options(selectinload(Campaign.advocacies)) 

67 

68 result = await db.execute(query) 

69 campaigns = result.scalars().all() 

70 

71 # Convert to response models 

72 campaign_responses = [] 

73 for campaign in campaigns: 

74 campaign_dict = { 

75 **campaign.__dict__, 

76 "advocate_count": calculate_advocate_count(campaign), 

77 } 

78 campaign_responses.append(CampaignResponse(**campaign_dict)) 

79 

80 return CampaignListResponse( 

81 campaigns=campaign_responses, 

82 total=total, 

83 page=page, 

84 per_page=per_page, 

85 ) 

86 

87 

88@router.get("/{campaign_id}", response_model=CampaignDetailResponse) 

89async def get_campaign( 

90 campaign_id: str, 

91 db: AsyncSession = Depends(get_db), 

92): 

93 """Get a single campaign by ID.""" 

94 query = select(Campaign).where(Campaign.id == campaign_id) 

95 query = query.options(selectinload(Campaign.advocacies)) 

96 

97 result = await db.execute(query) 

98 campaign = result.scalar_one_or_none() 

99 

100 if not campaign: 

101 raise HTTPException(status_code=404, detail="Campaign not found") 

102 

103 campaign_dict = { 

104 **campaign.__dict__, 

105 "advocate_count": calculate_advocate_count(campaign), 

106 } 

107 

108 return CampaignDetailResponse(**campaign_dict) 

109 

110 

111@router.post("", response_model=CampaignDetailResponse, status_code=201) 

112async def create_campaign( 

113 campaign_data: CampaignCreate, 

114 creator: Creator = Depends(get_required_creator), 

115 db: AsyncSession = Depends(get_db), 

116): 

117 """Create a new campaign.""" 

118 # Validate at least one wallet address 

119 if not any([ 

120 campaign_data.eth_wallet_address, 

121 campaign_data.btc_wallet_address, 

122 campaign_data.doge_wallet_address, 

123 campaign_data.sol_wallet_address, 

124 ]): 

125 raise HTTPException( 

126 status_code=400, 

127 detail="At least one wallet address (BTC, ETH, DOGE, or SOL) is required", 

128 ) 

129 

130 try: 

131 campaign_dict = campaign_data.model_dump(exclude={"contact_email"}) 

132 campaign = Campaign( 

133 **campaign_dict, 

134 creator_id=creator.id, 

135 contact_email=creator.email, # Use creator's email 

136 ) 

137 

138 db.add(campaign) 

139 await db.flush() # Flush to get campaign.id 

140 

141 # Create feed event for campaign creation 

142 from app.db.models import FeedEvent, FeedEventType 

143 feed_event = FeedEvent( 

144 event_type=FeedEventType.CAMPAIGN_CREATED, 

145 campaign_id=campaign.id, 

146 agent_id=None, 

147 event_metadata={"title": campaign.title}, 

148 ) 

149 db.add(feed_event) 

150 

151 await db.commit() 

152 await db.refresh(campaign) 

153 

154 # Load advocacies for response 

155 await db.refresh(campaign, ["advocacies"]) 

156 

157 campaign_dict = { 

158 **campaign.__dict__, 

159 "advocate_count": 0, 

160 } 

161 

162 return CampaignDetailResponse(**campaign_dict) 

163 except Exception as e: 

164 await db.rollback() 

165 raise HTTPException(status_code=500, detail=f"Failed to create campaign: {str(e)}") 

166 

167 

168@router.patch("/{campaign_id}", response_model=CampaignDetailResponse) 

169async def update_campaign( 

170 campaign_id: str, 

171 campaign_data: CampaignUpdate, 

172 creator: Creator = Depends(get_required_creator), 

173 db: AsyncSession = Depends(get_db), 

174): 

175 """Update a campaign (creator must own it).""" 

176 query = select(Campaign).where(Campaign.id == campaign_id) 

177 result = await db.execute(query) 

178 campaign = result.scalar_one_or_none() 

179 

180 if not campaign: 

181 raise HTTPException(status_code=404, detail="Campaign not found") 

182 

183 if campaign.creator_id != creator.id: 

184 raise HTTPException(status_code=403, detail="Not authorized to update this campaign") 

185 

186 try: 

187 # Update fields 

188 update_data = campaign_data.model_dump(exclude_unset=True) 

189 for field, value in update_data.items(): 

190 setattr(campaign, field, value) 

191 

192 await db.commit() 

193 await db.refresh(campaign) 

194 await db.refresh(campaign, ["advocacies"]) 

195 

196 campaign_dict = { 

197 **campaign.__dict__, 

198 "advocate_count": calculate_advocate_count(campaign), 

199 } 

200 

201 return CampaignDetailResponse(**campaign_dict) 

202 except Exception as e: 

203 await db.rollback() 

204 raise HTTPException(status_code=500, detail=f"Failed to update campaign: {str(e)}") 

205 

206 

207@router.delete("/{campaign_id}", status_code=204) 

208async def delete_campaign( 

209 campaign_id: str, 

210 creator: Creator = Depends(get_required_creator), 

211 db: AsyncSession = Depends(get_db), 

212): 

213 """Cancel/delete a campaign (creator must own it).""" 

214 query = select(Campaign).where(Campaign.id == campaign_id) 

215 result = await db.execute(query) 

216 campaign = result.scalar_one_or_none() 

217 

218 if not campaign: 

219 raise HTTPException(status_code=404, detail="Campaign not found") 

220 

221 if campaign.creator_id != creator.id: 

222 raise HTTPException(status_code=403, detail="Not authorized to delete this campaign") 

223 

224 try: 

225 campaign.status = CampaignStatus.CANCELLED 

226 await db.commit() 

227 return None 

228 except Exception as e: 

229 await db.rollback() 

230 raise HTTPException(status_code=500, detail=f"Failed to delete campaign: {str(e)}") 

231 

232 

233@router.get("/{campaign_id}/donations", response_model=DonationListResponse) 

234async def list_donations( 

235 campaign_id: str, 

236 page: int = Query(1, ge=1), 

237 per_page: int = Query(20, ge=1, le=100), 

238 chain: Optional[str] = Query(None, pattern="^(btc|eth|doge|sol)$"), 

239 db: AsyncSession = Depends(get_db), 

240): 

241 """List donations for a campaign.""" 

242 # Verify campaign exists 

243 campaign_result = await db.execute( 

244 select(Campaign).where(Campaign.id == campaign_id) 

245 ) 

246 campaign = campaign_result.scalar_one_or_none() 

247 if not campaign: 

248 raise HTTPException(status_code=404, detail="Campaign not found") 

249 

250 # Build query 

251 query = select(Donation).where(Donation.campaign_id == campaign_id) 

252 

253 if chain: 

254 query = query.where(Donation.chain == chain) 

255 

256 # Get total count 

257 count_query = select(func.count()).select_from(query.subquery()) 

258 total_result = await db.execute(count_query) 

259 total = total_result.scalar() or 0 

260 

261 # Apply pagination and ordering 

262 query = query.order_by(Donation.confirmed_at.desc()) 

263 query = query.offset((page - 1) * per_page).limit(per_page) 

264 

265 result = await db.execute(query) 

266 donations = result.scalars().all() 

267 

268 donation_responses = [DonationResponse(**donation.__dict__) for donation in donations] 

269 

270 return DonationListResponse( 

271 donations=donation_responses, 

272 total=total, 

273 page=page, 

274 per_page=per_page, 

275 )