Skip to content

Instantly share code, notes, and snippets.

@jgarvin
Created December 3, 2023 16:23
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save jgarvin/42539190ea136d92db0ee898daf8ac8b to your computer and use it in GitHub Desktop.
Save jgarvin/42539190ea136d92db0ee898daf8ac8b to your computer and use it in GitHub Desktop.
So python asyncio only has weak refs to tasks in the event loop...
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import asyncio
from asyncio import Task
import contextvars
from typing import Any, Coroutine, TypeVar
T = TypeVar("T")
ALL_TASKS: set[Task[Any]] = set()
original_create_task = asyncio.create_task
def sane_create_task(
coro: Coroutine[T, Any, Any],
*,
name: str | None = None,
context: contextvars.Context | None = None
) -> asyncio.Task[T]:
new_task = original_create_task(coro, name=name, context=context)
def remove_task_from_master_set(task: Task[Any]) -> None:
ALL_TASKS.remove(task)
new_task.add_done_callback(remove_task_from_master_set)
ALL_TASKS.add(new_task)
return new_task
asyncio.create_task = sane_create_task
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment