Coverage for src/cstlcore/newsletter/router.py: 35%

1from fastapi import APIRouter, BackgroundTasks, Depends 

2from sqlmodel import Session, select, func 

3 

4from cstlcore.database.dependencies import get_session 

5from cstlcore.newsletter.models import ( 

6 NewsletterForm, 

7 NewsletterSubscribed, 

8 NewsletterSubscribedCreate, 

9) 

10from cstlcore.newsletter.services import send_email_newsletter, verify_unsubscribe_token 

11from cstlcore.settings import settings 

12 

13router = APIRouter() 

14 

15 

16@router.post("/newsletter/send") 

17async def send_newsletter( 

18 newsletter_form: NewsletterForm, 

19 background_tasks: BackgroundTasks, 

20 session: Session = Depends(get_session), 

21): 

22 # Verify admin token 

23 if ( 

24 not newsletter_form.admin_token 

25 or newsletter_form.admin_token != settings.newsletter.admin_token 

26 ): 

27 return {"message": "Invalid admin token."} 

28 

29 # Get all subscribed emails from the newsletter subscription table 

30 subscriptions = session.exec(select(NewsletterSubscribed)).all() 

31 if not subscriptions: 

32 return {"message": "No one is subscribed to the newsletter."} 

33 

34 # Add the email sending task to background tasks 

35 background_tasks.add_task( 

36 send_email_newsletter, 

37 [sub.email for sub in subscriptions], 

38 newsletter_form.title, 

39 newsletter_form.content, 

40 f"{settings.services.current}/newsletter/unsubscribe", 

41 ) 

42 

43 return { 

44 "message": f"Newsletter queued for {len(subscriptions)} members. Emails will be sent in the background." 

45 } 

46 

47 

48@router.get("/newsletter/unsubscribe") 

49async def unsubscribe_newsletter(token: str, session: Session = Depends(get_session)): 

50 email_member = verify_unsubscribe_token(token) 

51 if not email_member: 

52 return {"message": "Invalid or expired token."} 

53 

54 # Remove the newsletter subscription 

55 subscription = session.exec( 

56 select(NewsletterSubscribed).where(NewsletterSubscribed.email == email_member) 

57 ).first() 

58 if subscription: 

59 session.delete(subscription) 

60 session.commit() 

61 

62 return {"ok": True} 

63 

64 

65@router.post("/newsletter/subscribe") 

66async def subscribe_newsletter( 

67 subscription: NewsletterSubscribedCreate, 

68 session: Session = Depends(get_session), 

69): 

70 # Check if email is already subscribed 

71 existing_subscription = session.exec( 

72 select(NewsletterSubscribed).where( 

73 NewsletterSubscribed.email == subscription.email 

74 ) 

75 ).first() 

76 

77 if existing_subscription: 

78 return {"message": "Email is already subscribed to the newsletter."} 

79 

80 # Create new subscription 

81 new_subscription = NewsletterSubscribed(email=subscription.email) 

82 session.add(new_subscription) 

83 session.commit() 

84 session.refresh(new_subscription) 

85 

86 return {"ok": True} 

87 

88 

89@router.get("/newsletter/subscribers/count") 

90async def get_subscribers_count( 

91 admin_token: str, 

92 session: Session = Depends(get_session), 

93): 

94 # Verify admin token 

95 if not admin_token or admin_token != settings.newsletter.admin_token: 

96 return {"message": "Invalid admin token."} 

97 

98 count = session.exec(select(func.count()).select_from(NewsletterSubscribed)).one() 

99 return {"count": count} 

100 

101 

102@router.get("/newsletter/subscribers") 

103async def get_subscribers_list( 

104 admin_token: str, 

105 session: Session = Depends(get_session), 

106): 

107 # Verify admin token 

108 if not admin_token or admin_token != settings.newsletter.admin_token: 

109 return {"message": "Invalid admin token."} 

110 

111 subscriptions = session.exec(select(NewsletterSubscribed)).all() 

112 return {"subscribers": [sub.email for sub in subscriptions], "count": len(subscriptions)}