Coverage for polar/campaign/service.py: 36%

1from datetime import UTC, datetime 

2 

3from sqlalchemy import select 

4 

5from polar.kit.services import ResourceServiceReader 

6from polar.models import Campaign, User 

7from polar.postgres import AsyncSession 

8 

9 

10class CampaignService(ResourceServiceReader[Campaign]): 

11 async def get_eligible(self, session: AsyncSession, user: User) -> Campaign | None: 

12 code = user.campaign_code 

13 if not code: 

14 return None 

15 

16 now = datetime.now(UTC) 

17 stmt = select(Campaign).where( 

18 Campaign.code == code, 

19 Campaign.starts_at <= now, 

20 Campaign.ends_at > now, 

21 Campaign.deleted_at.is_(None), 

22 ) 

23 res = await session.execute(stmt) 

24 campaign = res.scalars().unique().one_or_none() 

25 if not campaign: 

26 return None 

27 

28 # TODO: Check account threshold & user limits 

29 return campaign 

30 

31 

32campaign = CampaignService(Campaign)