from typing import Optional
import asyncio


def _current_task() -> Optional[asyncio.Task]:
  """Get the current running task, or `None` if no task is running."""
  # Note: `current_task` returns `None` if there is no current task, but it
  # throws if no currently running async loop.
  try:
    return asyncio.current_task()
  except RuntimeError:
    return None