Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e2e7882bfa | ||
|
|
4f9c2b9d5f | ||
|
|
5203bcf1ea | ||
|
|
f1e3bc8559 | ||
|
|
b97ca6f064 | ||
|
|
d1ea9874da |
@@ -1,7 +1,7 @@
|
|||||||
# MARK: Project info
|
# MARK: Project info
|
||||||
[project]
|
[project]
|
||||||
name = "corelibs"
|
name = "corelibs"
|
||||||
version = "0.18.2"
|
version = "0.20.0"
|
||||||
description = "Collection of utils for Python scripts"
|
description = "Collection of utils for Python scripts"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.13"
|
requires-python = ">=3.13"
|
||||||
|
|||||||
@@ -4,7 +4,12 @@ Various debug helpers
|
|||||||
|
|
||||||
import traceback
|
import traceback
|
||||||
import os
|
import os
|
||||||
|
import sys
|
||||||
|
from typing import Tuple, Type
|
||||||
|
from types import TracebackType
|
||||||
|
|
||||||
|
# _typeshed.OptExcInfo
|
||||||
|
OptExcInfo = Tuple[None, None, None] | Tuple[Type[BaseException], BaseException, TracebackType]
|
||||||
|
|
||||||
def call_stack(
|
def call_stack(
|
||||||
start: int = 0,
|
start: int = 0,
|
||||||
@@ -41,4 +46,30 @@ def call_stack(
|
|||||||
# print(f"* HERE: {dump_data(stack)}")
|
# print(f"* HERE: {dump_data(stack)}")
|
||||||
return f"{separator}".join(f"{os.path.basename(f.filename)}:{f.name}:{f.lineno}" for f in __stack)
|
return f"{separator}".join(f"{os.path.basename(f.filename)}:{f.name}:{f.lineno}" for f in __stack)
|
||||||
|
|
||||||
|
|
||||||
|
def exception_stack(
|
||||||
|
exc_stack: OptExcInfo | None = None,
|
||||||
|
separator: str = ' -> '
|
||||||
|
) -> str:
|
||||||
|
"""
|
||||||
|
Exception traceback, if no sys.exc_info is set, run internal
|
||||||
|
|
||||||
|
Keyword Arguments:
|
||||||
|
exc_stack {OptExcInfo | None} -- _description_ (default: {None})
|
||||||
|
separator {str} -- _description_ (default: {' -> '})
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
str -- _description_
|
||||||
|
"""
|
||||||
|
if exc_stack is not None:
|
||||||
|
_, _, exc_traceback = exc_stack
|
||||||
|
else:
|
||||||
|
exc_traceback = None
|
||||||
|
_, _, exc_traceback = sys.exc_info()
|
||||||
|
stack = traceback.extract_tb(exc_traceback)
|
||||||
|
if not separator:
|
||||||
|
separator = ' -> '
|
||||||
|
# print(f"* HERE: {dump_data(stack)}")
|
||||||
|
return f"{separator}".join(f"{os.path.basename(f.filename)}:{f.name}:{f.lineno}" for f in stack)
|
||||||
|
|
||||||
# __END__
|
# __END__
|
||||||
|
|||||||
@@ -32,4 +32,6 @@ def jmespath_search(search_data: dict[Any, Any] | list[Any], search_params: str)
|
|||||||
raise ValueError(f"Type error for search_params: {excp}") from excp
|
raise ValueError(f"Type error for search_params: {excp}") from excp
|
||||||
return search_result
|
return search_result
|
||||||
|
|
||||||
|
# TODO: compile jmespath setup
|
||||||
|
|
||||||
# __END__
|
# __END__
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ from pathlib import Path
|
|||||||
from typing import MutableMapping, TextIO, TypedDict, Any, TYPE_CHECKING, cast
|
from typing import MutableMapping, TextIO, TypedDict, Any, TYPE_CHECKING, cast
|
||||||
from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel
|
from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel
|
||||||
from corelibs.string_handling.text_colors import Colors
|
from corelibs.string_handling.text_colors import Colors
|
||||||
from corelibs.debug_handling.debug_helpers import call_stack
|
from corelibs.debug_handling.debug_helpers import call_stack, exception_stack
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from multiprocessing import Queue
|
from multiprocessing import Queue
|
||||||
@@ -225,11 +225,13 @@ class LogParent:
|
|||||||
if extra is None:
|
if extra is None:
|
||||||
extra = {}
|
extra = {}
|
||||||
extra['stack_trace'] = call_stack(skip_last=2)
|
extra['stack_trace'] = call_stack(skip_last=2)
|
||||||
|
extra['exception_trace'] = exception_stack()
|
||||||
# write to console first with extra flag for filtering in file
|
# write to console first with extra flag for filtering in file
|
||||||
if log_error:
|
if log_error:
|
||||||
self.logger.log(
|
self.logger.log(
|
||||||
LoggingLevel.ERROR.value,
|
LoggingLevel.ERROR.value,
|
||||||
f"<=EXCEPTION> {msg}", *args, extra=dict(extra) | {'console': True}, stacklevel=2
|
f"<=EXCEPTION={extra['exception_trace']}> {msg} [{extra['stack_trace']}]",
|
||||||
|
*args, extra=dict(extra) | {'console': True}, stacklevel=2
|
||||||
)
|
)
|
||||||
self.logger.log(LoggingLevel.EXCEPTION.value, msg, *args, exc_info=True, extra=extra, stacklevel=2)
|
self.logger.log(LoggingLevel.EXCEPTION.value, msg, *args, exc_info=True, extra=extra, stacklevel=2)
|
||||||
|
|
||||||
|
|||||||
20
src/corelibs/requests_handling/auth_helpers.py
Normal file
20
src/corelibs/requests_handling/auth_helpers.py
Normal file
@@ -0,0 +1,20 @@
|
|||||||
|
"""
|
||||||
|
Various HTTP auth helpers
|
||||||
|
"""
|
||||||
|
|
||||||
|
from base64 import b64encode
|
||||||
|
|
||||||
|
|
||||||
|
def basic_auth(username: str, password: str) -> str:
|
||||||
|
"""
|
||||||
|
setup basic auth, for debug
|
||||||
|
|
||||||
|
Arguments:
|
||||||
|
username {str} -- _description_
|
||||||
|
password {str} -- _description_
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
str -- _description_
|
||||||
|
"""
|
||||||
|
token = b64encode(f"{username}:{password}".encode('utf-8')).decode("ascii")
|
||||||
|
return f'Basic {token}'
|
||||||
52
test-run/json_handling/jmespath_helper.py
Normal file
52
test-run/json_handling/jmespath_helper.py
Normal file
@@ -0,0 +1,52 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
|
||||||
|
"""
|
||||||
|
jmes path testing
|
||||||
|
"""
|
||||||
|
|
||||||
|
from corelibs.debug_handling.dump_data import dump_data
|
||||||
|
from corelibs.json_handling.jmespath_helper import jmespath_search
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
"""
|
||||||
|
Comment
|
||||||
|
"""
|
||||||
|
__set = {
|
||||||
|
'a': 'b',
|
||||||
|
'foobar': [1, 2, 'a'],
|
||||||
|
'bar': {
|
||||||
|
'a': 1,
|
||||||
|
'b': 'c'
|
||||||
|
},
|
||||||
|
'baz': [
|
||||||
|
{
|
||||||
|
'aa': 1,
|
||||||
|
'ab': 'cc'
|
||||||
|
},
|
||||||
|
{
|
||||||
|
'ba': 2,
|
||||||
|
'bb': 'dd'
|
||||||
|
},
|
||||||
|
],
|
||||||
|
'foo': {
|
||||||
|
'a': [1, 2, 3],
|
||||||
|
'b': ['a', 'b', 'c']
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
__get = [
|
||||||
|
'a',
|
||||||
|
'bar.a',
|
||||||
|
'foo.a',
|
||||||
|
'baz[].aa'
|
||||||
|
]
|
||||||
|
for __jmespath in __get:
|
||||||
|
result = jmespath_search(__set, __jmespath)
|
||||||
|
print(f"GET {__jmespath}: {dump_data(result)}")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
|
|
||||||
|
# __END__
|
||||||
@@ -3,9 +3,11 @@ Log logging_handling.log testing
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
# import atexit
|
# import atexit
|
||||||
|
import sys
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
# this is for testing only
|
# this is for testing only
|
||||||
from corelibs.logging_handling.log import Log, Logger
|
from corelibs.logging_handling.log import Log, Logger
|
||||||
|
from corelibs.debug_handling.debug_helpers import exception_stack, call_stack
|
||||||
from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel
|
from corelibs.logging_handling.logging_level_handling.logging_level import LoggingLevel
|
||||||
|
|
||||||
|
|
||||||
@@ -78,6 +80,8 @@ def main():
|
|||||||
__test = 5 / 0
|
__test = 5 / 0
|
||||||
print(f"Divied: {__test}")
|
print(f"Divied: {__test}")
|
||||||
except ZeroDivisionError as e:
|
except ZeroDivisionError as e:
|
||||||
|
print(f"** sys.exec_info(): {sys.exc_info()}")
|
||||||
|
print(f"** sys.exec_info(): [{exception_stack()}] | [{exception_stack(sys.exc_info())}] | [{call_stack()}]")
|
||||||
log.logger.critical("Divison through zero: %s", e)
|
log.logger.critical("Divison through zero: %s", e)
|
||||||
log.exception("Divison through zero: %s", e)
|
log.exception("Divison through zero: %s", e)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user