Coverage for polar/organization_access_token/tasks.py: 67%

1import uuid 

2from datetime import UTC, datetime 

3 

4from polar.worker import AsyncSessionMaker, TaskPriority, actor 

5 

6from .repository import OrganizationAccessTokenRepository 

7 

8 

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

10async def record_usage( 

11 organization_access_token_id: uuid.UUID, last_used_at: float 

12) -> None: 

13 async with AsyncSessionMaker() as session: 

14 repository = OrganizationAccessTokenRepository.from_session(session) 

15 await repository.record_usage( 

16 organization_access_token_id, datetime.fromtimestamp(last_used_at, tz=UTC) 

17 )