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

1import uuid 1 ctx1a

2from datetime import UTC, datetime 1 ctx1a

3 

4from polar.worker import AsyncSessionMaker, TaskPriority, actor 1 ctx1a

5 

6from .repository import OrganizationAccessTokenRepository 1 ctx1a

7 

8 

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

10async def record_usage( 1 ctx1a

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 )