Coverage for app / api / routes / feed.py: 100%

1from typing import Optional, List 

2from fastapi import APIRouter, Depends, Query 

3from sqlalchemy.ext.asyncio import AsyncSession 

4from sqlalchemy import select, func 

5 

6from app.db.database import get_db 

7from app.db.models import FeedEvent, FeedEventType 

8from app.schemas.feed import FeedEventResponse, FeedListResponse 

9from app.db.models import Campaign, Agent 

10 

11router = APIRouter() 

12 

13 

14@router.get("", response_model=FeedListResponse) 

15async def get_feed( 

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

17 per_page: int = Query(50, ge=1, le=100), 

18 filter_type: Optional[str] = Query(None, alias="filter", pattern="^(all|campaigns|advocacy|discussions)$"), 

19 db: AsyncSession = Depends(get_db), 

20): 

21 """Get activity feed.""" 

22 query = select(FeedEvent) 

23 

24 # Apply filter 

25 if filter_type == "campaigns": 

26 query = query.where(FeedEvent.event_type == FeedEventType.CAMPAIGN_CREATED) 

27 elif filter_type == "advocacy": 

28 query = query.where( 

29 FeedEvent.event_type.in_([ 

30 FeedEventType.ADVOCACY_ADDED, 

31 FeedEventType.ADVOCACY_STATEMENT, 

32 ]) 

33 ) 

34 elif filter_type == "discussions": 

35 query = query.where(FeedEvent.event_type == FeedEventType.WARROOM_POST) 

36 

37 # Get total count 

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

39 total_result = await db.execute(count_query) 

40 total = total_result.scalar() or 0 

41 

42 # Order by newest first 

43 query = query.order_by(FeedEvent.created_at.desc()) 

44 

45 # Apply pagination 

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

47 

48 result = await db.execute(query) 

49 events = result.scalars().all() 

50 

51 # Build response with enriched data 

52 event_responses = [] 

53 for event in events: 

54 campaign_title = None 

55 agent_name = None 

56 agent_avatar_url = None 

57 

58 if event.campaign_id: 

59 campaign_query = select(Campaign).where(Campaign.id == event.campaign_id) 

60 campaign_result = await db.execute(campaign_query) 

61 campaign = campaign_result.scalar_one_or_none() 

62 if campaign: 

63 campaign_title = campaign.title 

64 

65 if event.agent_id: 

66 agent_query = select(Agent).where(Agent.id == event.agent_id) 

67 agent_result = await db.execute(agent_query) 

68 agent = agent_result.scalar_one_or_none() 

69 if agent: 

70 agent_name = agent.name 

71 agent_avatar_url = agent.avatar_url 

72 

73 event_responses.append(FeedEventResponse( 

74 id=event.id, 

75 event_type=event.event_type, 

76 campaign_id=event.campaign_id, 

77 campaign_title=campaign_title, 

78 agent_id=event.agent_id, 

79 agent_name=agent_name, 

80 agent_avatar_url=agent_avatar_url, 

81 metadata=event.event_metadata, 

82 created_at=event.created_at, 

83 )) 

84 

85 return FeedListResponse( 

86 events=event_responses, 

87 total=total, 

88 page=page, 

89 per_page=per_page, 

90 )