fizzbuzz.py, async version

from __future__ import annotations

import asyncio
import time
from functools import partial
from itertools import count
from typing import Callable, AsyncGenerator

OUTPUT: list[int | str] = []


async def safe_sleeps(after: float, every: float) -> AsyncGenerator[int, None]:
    """Prevent sleep() drifting (especially with Windows 15mS clock precision)"""
    start: int = time.time_ns()
    for n in count():
        tick = start + int((after + every * n) * (1_000_000_000))
        while time.time_ns() < tick:
            await asyncio.sleep(0.01)
        yield tick


async def after_every(cb: Callable[[], None], after: float, every: float) -> None:
    async for _ in safe_sleeps(after, every):
        cb()


def digits(counter: Callable[[], int]) -> None:
    print(*OUTPUT[-2:], sep="", flush=True)
    OUTPUT[:] = (counter(), "")


async def main() -> None:
    asyncio.create_task(after_every(partial(OUTPUT.append, "Fizz"), 2.1, 3))
    asyncio.create_task(after_every(partial(OUTPUT.append, "Buzz"), 4.2, 5))
    await after_every(partial(digits, partial(next, count(start=1))), 0, 1)


try:
    asyncio.run(main())
except KeyboardInterrupt:
    pass