delta-barth/delta-barth-py
1
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects 3 Releases 3 Wiki Activity

base structure for logging and logging management via session, closes #2

Browse Source
This commit is contained in:
Florian Förster
2025-03-28 11:31:27 +01:00
parent 302ccc16db
commit d1d665e60a
9 changed files with 128 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
src/delta_barth/errors.py
View File

@@ -6,7 +6,7 @@ from functools import wraps
from typing import Any, Final
from delta_barth.constants import DEFAULT_API_ERR_CODE, DEFAULT_INTERNAL_ERR_CODE
from delta_barth.logging import logger_wrapped_results as logger
from delta_barth.logging import logger_status, logger_wrapped_results
from delta_barth.types import DataPipeStates, Status
if t.TYPE_CHECKING:
@@ -151,23 +151,32 @@ class StatusHandler:
state: Status,
) -> None:
if state == self.SUCCESS:
logger_status.info(
"[STATUS] Raise for status - SUCCESS. all good.", stack_info=True
)
return
code = state.code
descr = state.description
msg = state.message
exc: Exception
if code < DEFAULT_INTERNAL_ERR_CODE:
raise _construct_exception(UDataProcessingError, descr, msg)
exc = _construct_exception(UDataProcessingError, descr, msg)
elif DEFAULT_INTERNAL_ERR_CODE <= code < DEFAULT_API_ERR_CODE:
raise _construct_exception(UInternalError, descr, msg)
exc = _construct_exception(UInternalError, descr, msg)
else:
api_err = state.api_server_error
assert api_err is not None, (
"error code inidcated API error, but no error instance found"
)
add_info = api_err.model_dump(exclude_none=True)
raise _construct_exception(UApiError, descr, msg, add_info)
exc = _construct_exception(UApiError, descr, msg, add_info)
logger_status.error(
"[STATUS] Raise for status - Error occurred: %s", exc, stack_info=True
)
raise exc
STATUS_HANDLER: Final[StatusHandler] = StatusHandler()
@@ -229,24 +238,24 @@ def wrap_result(
def wrap_result(func: Callable[P, T]) -> Callable[P, ResultWrapper[T]]:
@wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> ResultWrapper[T]:
status: ResultWrapper[T]
wrapped_result: ResultWrapper[T]
try:
res = func(*args, **kwargs)
status = ResultWrapper(
wrapped_result = ResultWrapper(
result=res, exception=None, code_on_error=code_on_error
)
except Exception as err:
status = ResultWrapper(
wrapped_result = ResultWrapper(
result=NotSet(), exception=err, code_on_error=code_on_error
)
logger.error(
"An exception in routine %s occurred - msg: %s, stack trace:",
logger_wrapped_results.info(
"[RESULT-WRAPPER] An exception in routine %s occurred - msg: %s, stack trace:",
func.__name__,
str(err),
stack_info=True,
)
return status
return wrapped_result
return wrapper