Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion logwrap/__init__.pxd
Original file line number Diff line number Diff line change
@@ -1 +1 @@
cdef tuple __all__
cpdef tuple __all__
5 changes: 2 additions & 3 deletions logwrap/__init__.pyx
Original file line number Diff line number Diff line change
@@ -1,10 +1,9 @@
from .repr_utils cimport PrettyFormat, PrettyRepr, PrettyStr
from .repr_utils import pretty_repr, pretty_str
from .repr_utils cimport PrettyFormat, PrettyRepr, PrettyStr, pretty_repr, pretty_str

from .log_wrap cimport LogWrap
from .log_wrap import logwrap, BoundParameter, bind_args_kwargs

cdef tuple __all__ = (
cpdef tuple __all__ = (
"LogWrap",
"logwrap",
"PrettyFormat",
Expand Down
5 changes: 3 additions & 2 deletions logwrap/class_decorator.pxd
Original file line number Diff line number Diff line change
Expand Up @@ -16,5 +16,6 @@


cdef class BaseDecorator:
cdef readonly object _func
cdef dict __dict__
cdef:
readonly object _func
dict __dict__
51 changes: 23 additions & 28 deletions logwrap/log_wrap.pxd
Original file line number Diff line number Diff line change
Expand Up @@ -20,39 +20,34 @@ import typing
from logwrap cimport class_decorator


cdef unsigned int indent
cdef:
unsigned int indent


cdef str pretty_repr(
src: typing.Any,
unsigned int indent=?,
bint no_indent_start=?,
unsigned int max_indent=?,
unsigned int indent_step=?
)
class LogWrap(class_decorator.BaseDecorator):
"""Base class for LogWrap implementation."""

cdef:
public unsigned int log_level
public unsigned int exc_level
public unsigned int max_indent

cdef class LogWrap(class_decorator.BaseDecorator):
"""Base class for LogWrap implementation."""
public bint log_call_args
public bint log_call_args_on_exc
public bint log_traceback
public bint log_result_obj

cdef public unsigned int log_level
cdef public unsigned int exc_level
cdef public unsigned int max_indent
readonly object _spec
readonly object _logger

cdef public bint log_call_args
cdef public bint log_call_args_on_exc
cdef public bint log_traceback
cdef public bint log_result_obj
list __blacklisted_names
list __blacklisted_exceptions

cdef readonly object _spec
cdef readonly object _logger
cpdef object pre_process_param(self, object arg)
cpdef str post_process_param(self, object arg, str arg_repr)

cdef list __blacklisted_names
cdef list __blacklisted_exceptions

cdef str _get_func_args_repr(
self, sig: inspect.Signature, tuple args, dict kwargs
)
cdef void _make_done_record(self, str func_name, result: typing.Any)
cdef void _make_calling_record(self, str name, str arguments, str method=?)
cdef void _make_exc_record(self, str name, str arguments)
cdef:
str _get_func_args_repr(self, sig: inspect.Signature, tuple args, dict kwargs)
void _make_done_record(self, str func_name, result: typing.Any)
void _make_calling_record(self, str name, str arguments, str method=?)
void _make_exc_record(self, str name, str arguments)
251 changes: 119 additions & 132 deletions logwrap/log_wrap.pyx
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ from logwrap cimport repr_utils
from logwrap cimport class_decorator


cdef tuple __all__ = ("LogWrap", "logwrap", "BoundParameter", "bind_args_kwargs")
cpdef tuple __all__ = ("LogWrap", "logwrap", "BoundParameter", "bind_args_kwargs")

logger = logging.getLogger("logwrap") # type: logging.Logger

Expand All @@ -36,19 +36,6 @@ fmt = "\n{spc:<{indent}}{{key!r}}={{val}},{{annotation}}".format(spc="", indent=
comment = "\n{spc:<{indent}}# {{kind!s}}:".format(spc="", indent=indent).format


cdef str pretty_repr(
src: typing.Any,
unsigned int indent=0,
bint no_indent_start=False,
unsigned int max_indent=20,
unsigned int indent_step=4
):
"""Make human readable repr of object."""
return repr_utils.PrettyRepr(max_indent=max_indent, indent_step=indent_step)(
src=src, indent=indent, no_indent_start=no_indent_start
)


class BoundParameter:
"""Parameter-like object store BOUND with value parameter."""

Expand Down Expand Up @@ -286,9 +273,7 @@ cdef class LogWrap(class_decorator.BaseDecorator):
"log_result_obj={self.log_result_obj}, )".format(cls=self.__class__.__name__, self=self, spec=self._spec)
)

def pre_process_param(
self, arg: BoundParameter
) -> typing.Union[BoundParameter, typing.Tuple[BoundParameter, typing.Any], None]:
cpdef object pre_process_param(self, object arg: BoundParameter):
"""Process parameter for the future logging.

:param arg: bound parameter
Expand All @@ -302,7 +287,7 @@ cdef class LogWrap(class_decorator.BaseDecorator):
"""
return arg

def post_process_param(self, arg: BoundParameter, str arg_repr: str) -> str:
cpdef str post_process_param(self, object arg: BoundParameter, str arg_repr):
"""Process parameter for the future logging.

:param arg: bound parameter
Expand All @@ -318,123 +303,125 @@ cdef class LogWrap(class_decorator.BaseDecorator):
"""
return arg_repr

cdef str _get_func_args_repr(self, sig: inspect.Signature, tuple args, dict kwargs):
"""Internal helper for reducing complexity of decorator code.

:param sig: function signature
:type sig: inspect.Signature
:param args: not keyworded arguments
:type args: typing.Tuple
:param kwargs: keyworded arguments
:type kwargs: typing.Dict[str, typing.Any]
:return: repr over function arguments
:rtype: str

.. versionchanged:: 3.3.0 Use pre- and post- processing of params during execution
"""
if not (self.log_call_args or self.log_call_args_on_exc):
return ""

cdef str param_str = ""
cdef str val
cdef str annotation

last_kind = None
for param in bind_args_kwargs(sig, *args, **kwargs):
if param.name in self.blacklisted_names:
continue

preprocessed = self.pre_process_param(param)
if preprocessed is None:
continue

if isinstance(preprocessed, (tuple, list)):
param, value = preprocessed
else:
value = param.value

if param.empty is value:
if param.VAR_POSITIONAL == param.kind:
value = ()
elif param.VAR_KEYWORD == param.kind:
value = {}

val = pretty_repr(src=value, indent=indent + 4, no_indent_start=True, max_indent=self.max_indent)

val = self.post_process_param(param, val)

if last_kind != param.kind:
param_str += comment(kind=param.kind)
last_kind = param.kind

if param.empty is param.annotation:
annotation = ""
else:
annotation = " # type: {param.annotation!s}".format(param=param)

param_str += fmt(key=param.name, annotation=annotation, val=val)
if param_str:
param_str += "\n"
return param_str

cdef void _make_done_record(self, str func_name, result: typing.Any):
"""Construct success record.

:type func_name: str
:type result: typing.Any
"""
cdef str msg = "Done: {name!r}".format(name=func_name)

if self.log_result_obj:
msg += " with result:\n{result}".format(
result=pretty_repr(
result,
indent=0,
no_indent_start=False,
max_indent=self.max_indent,
indent_step=4

cdef:
str _get_func_args_repr(self, sig: inspect.Signature, tuple args, dict kwargs):
"""Internal helper for reducing complexity of decorator code.

:param sig: function signature
:type sig: inspect.Signature
:param args: not keyworded arguments
:type args: typing.Tuple
:param kwargs: keyworded arguments
:type kwargs: typing.Dict[str, typing.Any]
:return: repr over function arguments
:rtype: str

.. versionchanged:: 3.3.0 Use pre- and post- processing of params during execution
"""
if not (self.log_call_args or self.log_call_args_on_exc):
return ""

cdef:
str param_str = ""
str val
str annotation

last_kind = None
for param in bind_args_kwargs(sig, *args, **kwargs):
if param.name in self.blacklisted_names:
continue

preprocessed = self.pre_process_param(param)
if preprocessed is None:
continue

if isinstance(preprocessed, (tuple, list)):
param, value = preprocessed
else:
value = param.value

if param.empty is value:
if param.VAR_POSITIONAL == param.kind:
value = ()
elif param.VAR_KEYWORD == param.kind:
value = {}

val = repr_utils.pretty_repr(src=value, indent=indent + 4, no_indent_start=True, max_indent=self.max_indent)

val = self.post_process_param(param, val)

if last_kind != param.kind:
param_str += comment(kind=param.kind)
last_kind = param.kind

if param.empty is param.annotation:
annotation = ""
else:
annotation = " # type: {param.annotation!s}".format(param=param)

param_str += fmt(key=param.name, annotation=annotation, val=val)
if param_str:
param_str += "\n"
return param_str

void _make_done_record(self, str func_name, result: typing.Any):
"""Construct success record.

:type func_name: str
:type result: typing.Any
"""
cdef str msg = "Done: {name!r}".format(name=func_name)

if self.log_result_obj:
msg += " with result:\n{result}".format(
result=repr_utils.pretty_repr(
result,
indent=0,
no_indent_start=False,
max_indent=self.max_indent,
indent_step=4

)
)
self._logger.log(level=self.log_level, msg=msg) # type: ignore

void _make_calling_record(self, str name, str arguments, str method="Calling"):
"""Make log record before execution.

:type name: str
:type arguments: str
:type method: str
"""
self._logger.log( # type: ignore
level=self.log_level,
msg="{method}: \n{name!r}({arguments})".format(
method=method, name=name, arguments=arguments if self.log_call_args else ""
),
)
self._logger.log(level=self.log_level, msg=msg) # type: ignore

cdef void _make_calling_record(self, str name, str arguments, str method="Calling"):
"""Make log record before execution.

:type name: str
:type arguments: str
:type method: str
"""
self._logger.log( # type: ignore
level=self.log_level,
msg="{method}: \n{name!r}({arguments})".format(
method=method, name=name, arguments=arguments if self.log_call_args else ""
),
)

cdef void _make_exc_record(self, str name, str arguments):
"""Make log record if exception raised.

:type name: str
:type arguments: str
"""
exc_info = sys.exc_info()
stack = traceback.extract_stack()
tb = traceback.extract_tb(exc_info[2])
full_tb = stack[:2] + tb # cut decorator and build full traceback
exc_line = traceback.format_exception_only(*exc_info[:2])
# Make standard traceback string
cdef str tb_text = "Traceback (most recent call last):\n" + "".join(traceback.format_list(full_tb)) + "".join(exc_line)

self._logger.log( # type: ignore
level=self.exc_level,
msg="Failed: \n{name!r}({arguments})\n{tb_text}".format(
name=name,
arguments=arguments if self.log_call_args_on_exc else "",
tb_text=tb_text if self.log_traceback else "",
),
exc_info=False,
)
void _make_exc_record(self, str name, str arguments):
"""Make log record if exception raised.

:type name: str
:type arguments: str
"""
exc_info = sys.exc_info()
stack = traceback.extract_stack()
tb = traceback.extract_tb(exc_info[2])
full_tb = stack[:2] + tb # cut decorator and build full traceback
exc_line = traceback.format_exception_only(*exc_info[:2])
# Make standard traceback string
cdef str tb_text = "Traceback (most recent call last):\n" + "".join(traceback.format_list(full_tb)) + "".join(exc_line)

self._logger.log( # type: ignore
level=self.exc_level,
msg="Failed: \n{name!r}({arguments})\n{tb_text}".format(
name=name,
arguments=arguments if self.log_call_args_on_exc else "",
tb_text=tb_text if self.log_traceback else "",
),
exc_info=False,
)

def _get_function_wrapper(self, func: typing.Callable) -> typing.Callable:
"""Here should be constructed and returned real decorator.
Expand Down
Loading