Coverage for polar/meter/tasks.py: 53%

1import uuid 

2 

3from apscheduler.triggers.cron import CronTrigger 

4from sqlalchemy.orm import joinedload 

5 

6from polar.exceptions import PolarTaskError 

7from polar.meter.repository import MeterRepository 

8from polar.meter.service import meter as meter_service 

9from polar.models import Meter 

10from polar.worker import AsyncSessionMaker, TaskPriority, actor 

11 

12 

13class MeterTaskError(PolarTaskError): ... 13 ↛ 16line 13 didn't jump to line 16 because

14 

15 

16class MeterDoesNotExist(MeterTaskError): 

17 def __init__(self, meter_id: uuid.UUID) -> None: 

18 self.meter_id = meter_id 

19 message = f"The meter with id {meter_id} does not exist." 

20 super().__init__(message) 

21 

22 

23@actor( 

24 actor_name="meter.enqueue_billing", 

25 cron_trigger=CronTrigger.from_crontab("*/5 * * * *"), 

26 priority=TaskPriority.LOW, 

27) 

28async def meter_enqueue_billing() -> None: 

29 async with AsyncSessionMaker() as session: 

30 await meter_service.enqueue_billing(session) 

31 

32 

33@actor(actor_name="meter.billing_entries", priority=TaskPriority.LOW) 

34async def meter_billing_entries(meter_id: uuid.UUID) -> None: 

35 async with AsyncSessionMaker() as session: 

36 repository = MeterRepository.from_session(session) 

37 meter = await repository.get_by_id( 

38 meter_id, options=(joinedload(Meter.last_billed_event),) 

39 ) 

40 if meter is None: 

41 raise MeterDoesNotExist(meter_id) 

42 

43 await meter_service.create_billing_entries(session, meter)