Add traceback property to AnyIOTaskResult class

This commit is contained in:
Gabriel Luiz Freitas Almeida 2023-12-18 10:50:01 -03:00
commit 2edacce305

View file

@ -1,8 +1,11 @@
import traceback
from typing import Any, Callable, Optional, Tuple
import anyio
from langflow.services.task.backends.base import TaskBackend
from loguru import logger
from langflow.services.task.backends.base import TaskBackend
class AnyIOTaskResult:
def __init__(self, scope):
@ -17,6 +20,12 @@ class AnyIOTaskResult:
return "FAILURE" if self._exception is not None else "SUCCESS"
return self._status
@property
def traceback(self) -> str:
if self._traceback is not None:
return "".join(traceback.format_tb(self._traceback))
return ""
@property
def result(self) -> Any:
return self._result
@ -29,6 +38,7 @@ class AnyIOTaskResult:
self._result = await func(*args, **kwargs)
except Exception as e:
self._exception = e
self._traceback = e.__traceback__
finally:
self._status = "DONE"