Garmelon/PFERD
1
0
Fork 0
mirror of https://github.com/Garmelon/PFERD.git synced 2026-04-12 07:25:04 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Save stacktraces in context manager and print on error

Browse source
This commit is contained in:
I-Al-Istannen 2025-12-09 17:01:24 +01:00
parent e246053de2
commit 4eab927899
1 changed files with 13 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

13
PFERD/utils.py
View file

@ -2,6 +2,7 @@ import asyncio
import getpass import getpass
import sys import sys
import threading import threading
import traceback
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from collections.abc import Callable from collections.abc import Callable
from contextlib import AsyncExitStack from contextlib import AsyncExitStack
@ -114,13 +115,23 @@ class ReusableAsyncContextManager(ABC, Generic[T]):
def __init__(self) -> None: def __init__(self) -> None:
self._active = False self._active = False
self._stack = AsyncExitStack() self._stack = AsyncExitStack()
self._create_stacktrace = traceback.format_stack()
self._enter_stacktraces = []
@abstractmethod @abstractmethod
async def _on_aenter(self) -> T: async def _on_aenter(self) -> T:
pass pass
async def __aenter__(self) -> T: async def __aenter__(self) -> T:
self._enter_stacktraces.append(traceback.format_stack())
if self._active: if self._active:
print("Context manager was already active. Created at:")
print("".join(self._create_stacktrace))
print("\n== Previous __aenter__ calls")
for i, stacktrace in enumerate(self._enter_stacktraces, start=1):
print(f"\n-- __aenter__ call #{i} at:")
print("".join(stacktrace))
raise RuntimeError("Nested or otherwise concurrent usage is not allowed") raise RuntimeError("Nested or otherwise concurrent usage is not allowed")
self._active = True self._active = True
@ -144,6 +155,8 @@ class ReusableAsyncContextManager(ABC, Generic[T]):
if not self._active: if not self._active:
raise RuntimeError("__aexit__ called too many times") raise RuntimeError("__aexit__ called too many times")
self._enter_stacktraces.pop()
result = await self._stack.__aexit__(exc_type, exc_value, traceback) result = await self._stack.__aexit__(exc_type, exc_value, traceback)
self._active = False self._active = False
return result return result