A python library that I am working on but encoutered troubles

output with python 3.12

uv run --python 3.12 tests.py                                                                                                                                                git:master*
Using CPython 3.12.9
Removed virtual environment at: .venv
Creating virtual environment at: .venv
INFO:root:[asyncio-task-ancestors] Patching asyncio task factory function!
DEBUG:root:[asyncio-task-ancestors] wrap asyncio.Task class!
----init----
New task name: Task-2
New task id: 130108021592064
----done----
[Test1] task: <Task pending name='Task-2' coro=<test1_subroutine() running at /home/control/asyncio-task-ancestors/tests.py:12> cb=[Task.task_wakeup()]>
[Test1] task id: 130108021592064
[Test1][Task=Task-2] has 1 ancestors (['Task-1'])
----init----
New task name: Task-3
New task id: 130108021592448
----done----
----init----
New task name: Task-4
New task id: 130108021592448
----done----

output with python 3.13

uv run --python 3.13 tests.py                                                                                                                                                git:master*
Using CPython 3.13.2
Removed virtual environment at: .venv
Creating virtual environment at: .venv
INFO:root:[asyncio-task-ancestors] Patching asyncio task factory function!
DEBUG:root:[asyncio-task-ancestors] wrap asyncio.Task class!
----init----
New task name: Task-2
New task id: 128326840481088
----done----
[Test1] task: <Task pending name='None' coro=<test1_subroutine() running at /home/control/asyncio-task-ancestors/tests.py:12> cb=[Task.task_wakeup()]>
[Test1] task id: 128326840481088
[Test1][Task=None] has 1 ancestors (['Task-1'])
----init----
New task name: Task-3
New task id: 128326840480912
----done----
----init----
New task name: Task-4
New task id: 128326840480912
----done----

output with python 3.9

uv run --python 3.9 tests.py                                                                                                                                                 git:master*
Using CPython 3.9.21
Removed virtual environment at: .venv
Creating virtual environment at: .venv
INFO:root:[asyncio-task-ancestors] Patching asyncio task factory function!
DEBUG:root:[asyncio-task-ancestors] wrap asyncio.Task class!
----init----
New task name: Task-2
New task id: 134885533998192
----done----
[Test1] task: <Task pending name='Task-2' coro=<test1_subroutine() running at /home/control/asyncio-task-ancestors/tests.py:12> cb=[<TaskWakeupMethWrapper object at 0x7aad7c4adb80>()]>
[Test1] task id: 134885533998192
[Test1][Task=Task-2] has 1 ancestors (['Task-1'])
----init----
New task name: Task-3
New task id: 134885541214720
----done----
----init----
New task name: Task-4
New task id: 134885541214720
----done----

patch_asyncio.py

import asyncio
import logging
import copy

def task_creation_wrapper(func):
    def wrapper(*args, **kwargs):
        try:
            current_task = asyncio.current_task()
        except:
            current_task = None
        new_task = func(*args, **kwargs)
        if current_task:
            try:
                new_task.ancestors = current_task.ancestors.copy()
            except:
                new_task.ancestors = []
            new_task.ancestors.append(current_task)
        else:
            new_task.ancestors = []
        return new_task
    return wrapper

def task_creation_wrapper_2(self, *args, **kwargs):
    try:
        current_task = asyncio.current_task()
    except:
        current_task = None
    print("----init----")
    new_task = asyncio.Task(*args, **kwargs)
    print(f"New task name: {new_task.get_name()}")
    print(f"New task id: {id(new_task)}")
    if current_task:
        try:
            new_task.ancestors = current_task.ancestors.copy()
        except:
            new_task.ancestors = []
        new_task.ancestors.append(current_task)
    else:
        new_task.ancestors = []
    print("----done----")
    return new_task

def patch():
    loop = asyncio.get_running_loop()
    logging.info("[asyncio-task-ancestors] Patching asyncio task factory function!")
    try:
        asyncio.current_task().ancestors = []
    except:
        pass
    if func := loop.get_task_factory():
        logging.debug("[asyncio-task-ancestors] Patch existing factory function!")
        loop.set_task_factory(task_creation_wrapper(func))
    else:
        logging.debug("[asyncio-task-ancestors] wrap asyncio.Task class!")
        loop.set_task_factory(task_creation_wrapper_2)

tests.py

import patch_asyncio
import asyncio
import inspect
import logging

async def test1():
    new_task = asyncio.create_task(test1_subroutine())
    await new_task

async def test1_subroutine():
    current_task = asyncio.current_task()
    print(f"[Test1] task: {current_task}")
    print(f"[Test1] task id: {id(current_task)}")
    ancestor_names = [x.get_name() for x in current_task.ancestors]
    print(f"[Test1][Task={current_task.get_name()}] has {len(current_task.ancestors)} ancestors ({ancestor_names})")

async def test2():
    await test2_subroutine1()
    
async def test2_subroutine1():
    print(inspect.getfile(asyncio.gather))
    await asyncio.gather(
        test2_subroutine2(),
        test2_subroutine2(),
        test2_subroutine2()
    )
    
async def test2_subroutine2():
    current_task = asyncio.current_task()
    print(current_task)
    ancestor_names = [x.get_name() for x in current_task.ancestors]
    print(f"[Test2-2][Task={current_task.get_name()}] has {len(current_task.ancestors)} ancestors ({ancestor_names})")
    async with asyncio.TaskGroup() as tg:
        tg.create_task(test2_subroutine3())
    
async def test2_subroutine3():
    current_task = asyncio.current_task()
    print(current_task)
    ancestor_names = [x.get_name() for x in current_task.ancestors]
    print(f"[Test2-3][Task={current_task.get_name()}] has {len(current_task.ancestors)} ancestors ({ancestor_names})")

async def main():
    logging.basicConfig(level=logging.DEBUG)
    patch_asyncio.patch()
    await test1()
    #await test2()

asyncio.run(main())