Coverage for polar/personal_access_token/tasks.py: 75%

1import uuid 

2from datetime import UTC, datetime 

3 

4from polar.worker import AsyncSessionMaker, TaskPriority, actor 

5 

6from .service import personal_access_token as personal_access_token_service 

7 

8 

9@actor(actor_name="personal_access_token.record_usage", priority=TaskPriority.LOW) 

10async def record_usage( 

11 personal_access_token_id: uuid.UUID, last_used_at: float 

12) -> None: 

13 async with AsyncSessionMaker() as session: 

14 await personal_access_token_service.record_usage( 

15 session, 

16 personal_access_token_id, 

17 datetime.fromtimestamp(last_used_at, tz=UTC), 

18 )