print progress information while waiting for a number of async tasks to finish

taskgathering.py

#!/usr/bin/env python3

# This is free and unencumbered software released into the public domain.
# Please refer to the terms of the "Unlicense" at https://spdx.org/licenses/Unlicense.html

# demo: https://asciinema.org/a/FubNfdJELHAL80xcAJl0RA7OW

import asyncio, random

# random time with some gauss distribution
rtime = lambda: max(random.gauss(10, 4), 1)

# wrap a single async task
async def spin(name, tasks, fn):
  tasks.append(name)
  try:
    result = await fn
    return result
  finally:
    tasks.remove(name)
    print(f"\033[2K\rDone: {name}")

# pointless work
async def sleeper(s):
  await asyncio.sleep(s)
  return f"slept for {s} seconds"

# task which waits for all the others to finish
# and prints some progress / status information
async def printer(spinner, total):
  step = 0
  index = 0
  while len(spinner):

    # clear line
    print("\033[2K\r", end="")
    
    # pick next index
    if step >= 5:
      index += 1
      step = 0
    index %= len(spinner)

    # calc done and print line
    done = total - len(spinner)
    print(f"({done:02d}/{total:02d}) {spinner[index]} ...", end="", flush=True)
    step += 1
    await asyncio.sleep(0.2)

  # print final newline
  print(f"\033[2K\r({total:02d}/{total:02d}) Finished")
  return "printer done"

# main tasks that launches some pointless work
# and waits for the results
async def main():

  progress = []
  work = []
  for i in range(10):
    work.append(spin(f"task-{i}", progress, sleeper(rtime())))
  result = await asyncio.gather(*work, printer(progress, len(work)))
  print(result)

# let's go
asyncio.run(main())