Coverage for async_durable_execution/_operation/replay_safe.py: 96%

1"""Replay-safe helpers for common non-deterministic values.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import random as _random 

7import time as _time 

8import uuid as _uuid 

9from datetime import datetime, timezone 

10 

11from .step import step 

12 

13 

14async def _random_value() -> float: 

15 return _random.random() 

16 

17 

18async def _now_value() -> datetime: 

19 return datetime.now(tz=timezone.utc) 

20 

21 

22async def _timestamp_value() -> float: 

23 return _time.time() 

24 

25 

26async def _uuid_value() -> _uuid.UUID: 

27 return _uuid.uuid4() 

28 

29 

30def random(*, name: str | None = None) -> asyncio.Task[float]: 

31 """Return a checkpointed `random.random()` value.""" 

32 return step(_random_value, name=name or "random") 

33 

34 

35def now(*, name: str | None = None) -> asyncio.Task[datetime]: 

36 """Return a checkpointed timezone-aware UTC `datetime`.""" 

37 return step(_now_value, name=name or "now") 

38 

39 

40def timestamp(*, name: str | None = None) -> asyncio.Task[float]: 

41 """Return a checkpointed Unix timestamp in seconds.""" 

42 return step(_timestamp_value, name=name or "timestamp") 

43 

44 

45def uuid(*, name: str | None = None) -> asyncio.Task[_uuid.UUID]: 

46 """Return a checkpointed UUID4 value.""" 

47 return step(_uuid_value, name=name or "uuid")