Provides a continuously updating stream of the current UTC time. This function generates a real-time feed of the current UTC timestamp, refreshing at a specified interval. Args: refresh_rate: The interval at which the current UTC time is refreshed. Defaults to
(
refresh_rate: datetime.timedelta = datetime.timedelta(seconds=60),
initial_delay: datetime.timedelta = datetime.timedelta(seconds=0),
)
| 40 | |
| 41 | @cache |
| 42 | def utc_now( |
| 43 | refresh_rate: datetime.timedelta = datetime.timedelta(seconds=60), |
| 44 | initial_delay: datetime.timedelta = datetime.timedelta(seconds=0), |
| 45 | ): |
| 46 | """ |
| 47 | Provides a continuously updating stream of the current UTC time. |
| 48 | |
| 49 | This function generates a real-time feed of the current UTC timestamp, refreshing |
| 50 | at a specified interval. |
| 51 | |
| 52 | Args: |
| 53 | refresh_rate: The interval at which the current |
| 54 | UTC time is refreshed. Defaults to 60 seconds. |
| 55 | |
| 56 | Returns: |
| 57 | A table containing a stream of the current UTC timestamps, updated |
| 58 | according to the specified refresh rate. |
| 59 | """ |
| 60 | return io.python.read( |
| 61 | TimestampSubject(refresh_rate=refresh_rate, initial_delay=initial_delay), |
| 62 | schema=TimestampSchema, |
| 63 | ) |
| 64 | |
| 65 | |
| 66 | def _get_now_timestamp_utc(for_test_only: pw.Pointer | None) -> pw.DateTimeUtc: |
no test coverage detected