콘텐츠로 이동

spakky-logging

구조화된 로깅 — Aspect 기반 자동 로깅, 컨텍스트 주입

Aspects

spakky.plugins.logging.aspects.logging_aspect

Logging aspects for automatic method call logging with masking and timing.

Provides sync and async AOP aspects that intercept methods annotated with @Logging and emit structured log messages including:

  • Method name and arguments (with sensitive data masking)
  • Return value (truncated to max_result_length)
  • Execution duration
  • Slow-call warnings when slow_threshold_ms is exceeded

AsyncLoggingAspect(config=None)

Bases: IAsyncAspect

Aspect for logging async method calls with execution time and data masking.

Initialize the async logging aspect.

Parameters:

Name Type Description Default
config LoggingConfig | None

Optional logging configuration. Uses defaults if None.

None
Source code in plugins/spakky-logging/src/spakky/plugins/logging/aspects/logging_aspect.py
def __init__(self, config: LoggingConfig | None = None) -> None:
    """Initialize the async logging aspect.

    Args:
        config: Optional logging configuration. Uses defaults if None.
    """
    self._config = config

around_async(joinpoint, *args, **kwargs) async

Log async method execution with timing and masking.

Parameters:

Name Type Description Default
joinpoint AsyncFunc

The async method being intercepted.

required
*args Any

Positional arguments to the method.

()
**kwargs Any

Keyword arguments to the method.

{}

Returns:

Type Description
Any

The result of the method execution.

Raises:

Type Description
Exception

Re-raises any exception after logging it.

Source code in plugins/spakky-logging/src/spakky/plugins/logging/aspects/logging_aspect.py
@Around(lambda x: Logged.exists(x) and iscoroutinefunction(x))
async def around_async(
    self,
    joinpoint: AsyncFunc,
    *args: Any,
    **kwargs: Any,  # noqa: ANN401
) -> Any:  # noqa: ANN401
    """Log async method execution with timing and masking.

    Args:
        joinpoint: The async method being intercepted.
        *args: Positional arguments to the method.
        **kwargs: Keyword arguments to the method.

    Returns:
        The result of the method execution.

    Raises:
        Exception: Re-raises any exception after logging it.
    """
    annotation: Logged = Logged.get(joinpoint)
    mask_keys = annotation.masking_keys or (
        self._config.mask_keys if self._config else DEFAULT_MASK_KEYS
    )
    mask = _build_mask(mask_keys)
    slow_threshold = annotation.slow_threshold_ms or (
        self._config.slow_threshold_ms
        if self._config
        else DEFAULT_SLOW_THRESHOLD_MS
    )
    max_result_length = annotation.max_result_length or (
        self._config.max_result_length
        if self._config
        else DEFAULT_MAX_RESULT_LENGTH
    )

    formatted_args = _format_args(args, kwargs) if annotation.log_args else "..."

    start: float = perf_counter()
    try:
        result = await joinpoint(*args, **kwargs)
    except Exception as e:
        elapsed_ms = (perf_counter() - start) * 1000
        error_msg = (
            f"[{type(self).__name__}] "
            f"{joinpoint.__qualname__}({formatted_args}) "
            f"raised {type(e).__name__} ({elapsed_ms:.2f}ms)"
        )
        logger.error(
            mask.sub(self.MASKING_TEXT, error_msg)
            if annotation.enable_masking
            else error_msg,
        )
        raise

    elapsed_ms = (perf_counter() - start) * 1000
    result_repr = (
        _truncate(repr(result), max_result_length)
        if annotation.log_result
        else "..."
    )
    msg = (
        f"[{type(self).__name__}] "
        f"{joinpoint.__qualname__}({formatted_args}) "
        f"-> {result_repr} ({elapsed_ms:.2f}ms)"
    )
    log_msg = mask.sub(self.MASKING_TEXT, msg) if annotation.enable_masking else msg

    if elapsed_ms >= slow_threshold:
        logger.log(WARNING, "[SLOW] %s", log_msg)
    else:
        logger.info(log_msg)

    return result

LoggingAspect(config=None)

Bases: IAspect

Aspect for logging synchronous method calls with execution time and data masking.

Initialize the sync logging aspect.

Parameters:

Name Type Description Default
config LoggingConfig | None

Optional logging configuration. Uses defaults if None.

None
Source code in plugins/spakky-logging/src/spakky/plugins/logging/aspects/logging_aspect.py
def __init__(self, config: LoggingConfig | None = None) -> None:
    """Initialize the sync logging aspect.

    Args:
        config: Optional logging configuration. Uses defaults if None.
    """
    self._config = config

around(joinpoint, *args, **kwargs)

Log sync method execution with timing and masking.

Parameters:

Name Type Description Default
joinpoint Func

The sync method being intercepted.

required
*args Any

Positional arguments to the method.

()
**kwargs Any

Keyword arguments to the method.

{}

Returns:

Type Description
Any

The result of the method execution.

Raises:

Type Description
Exception

Re-raises any exception after logging it.

Source code in plugins/spakky-logging/src/spakky/plugins/logging/aspects/logging_aspect.py
@Around(lambda x: Logged.exists(x) and not iscoroutinefunction(x))
def around(
    self,
    joinpoint: Func,
    *args: Any,
    **kwargs: Any,  # noqa: ANN401
) -> Any:  # noqa: ANN401
    """Log sync method execution with timing and masking.

    Args:
        joinpoint: The sync method being intercepted.
        *args: Positional arguments to the method.
        **kwargs: Keyword arguments to the method.

    Returns:
        The result of the method execution.

    Raises:
        Exception: Re-raises any exception after logging it.
    """
    annotation: Logged = Logged.get(joinpoint)
    mask_keys = annotation.masking_keys or (
        self._config.mask_keys if self._config else DEFAULT_MASK_KEYS
    )
    mask = _build_mask(mask_keys)
    slow_threshold = annotation.slow_threshold_ms or (
        self._config.slow_threshold_ms
        if self._config
        else DEFAULT_SLOW_THRESHOLD_MS
    )
    max_result_length = annotation.max_result_length or (
        self._config.max_result_length
        if self._config
        else DEFAULT_MAX_RESULT_LENGTH
    )

    formatted_args = _format_args(args, kwargs) if annotation.log_args else "..."

    start: float = perf_counter()
    try:
        result = joinpoint(*args, **kwargs)
    except Exception as e:
        elapsed_ms = (perf_counter() - start) * 1000
        error_msg = (
            f"[{type(self).__name__}] "
            f"{joinpoint.__qualname__}({formatted_args}) "
            f"raised {type(e).__name__} ({elapsed_ms:.2f}ms)"
        )
        logger.error(
            mask.sub(self.MASKING_TEXT, error_msg)
            if annotation.enable_masking
            else error_msg,
        )
        raise

    elapsed_ms = (perf_counter() - start) * 1000
    result_repr = (
        _truncate(repr(result), max_result_length)
        if annotation.log_result
        else "..."
    )
    msg = (
        f"[{type(self).__name__}] "
        f"{joinpoint.__qualname__}({formatted_args}) "
        f"-> {result_repr} ({elapsed_ms:.2f}ms)"
    )
    log_msg = mask.sub(self.MASKING_TEXT, msg) if annotation.enable_masking else msg

    if elapsed_ms >= slow_threshold:
        logger.log(WARNING, "[SLOW] %s", log_msg)
    else:
        logger.info(log_msg)

    return result

options: show_root_heading: false

Annotations

spakky.plugins.logging.annotation

Logging annotation for automatic method call logging.

Logged(enable_masking=True, masking_keys=list(), slow_threshold_ms=None, max_result_length=None, log_args=True, log_result=True) dataclass

Bases: FunctionAnnotation

Annotation for enabling automatic method logging.

Methods decorated with @Logging() will have their calls, arguments, return values, and execution time automatically logged.

Attributes:

Name Type Description
enable_masking bool

Whether to mask sensitive data in logs.

masking_keys list[str]

List of keys whose values should be masked. When empty, the global :attr:LoggingConfig.mask_keys is used.

slow_threshold_ms float | None

Per-method slow-call warning threshold. None falls back to :attr:LoggingConfig.slow_threshold_ms.

max_result_length int | None

Maximum repr length for the return value. None falls back to :attr:LoggingConfig.max_result_length.

log_args bool

Whether to include arguments in the log message.

log_result bool

Whether to include the return value in the log message.

enable_masking = True class-attribute instance-attribute

Whether to mask sensitive data in logs.

masking_keys = field(default_factory=list) class-attribute instance-attribute

Keys whose values should be masked. Empty = use global config.

slow_threshold_ms = None class-attribute instance-attribute

Per-method slow-call threshold (ms). None = use global config.

max_result_length = None class-attribute instance-attribute

Max repr length for return values. None = use global config.

log_args = True class-attribute instance-attribute

Whether to include arguments in log output.

log_result = True class-attribute instance-attribute

Whether to include the return value in log output.

logged(enable_masking=True, masking_keys=None, slow_threshold_ms=None, max_result_length=None, log_args=True, log_result=True)

Decorator for enabling automatic method logging.

Parameters:

Name Type Description Default
enable_masking bool

Whether to mask sensitive data in logs.

True
masking_keys list[str] | None

List of keys whose values should be masked. When empty, the global :attr:LoggingConfig.mask_keys is used.

None
slow_threshold_ms float | None

Per-method slow-call warning threshold. None falls back to :attr:LoggingConfig.slow_threshold_ms.

None
max_result_length int | None

Maximum repr length for the return value. None falls back to :attr:LoggingConfig.max_result_length.

None
log_args bool

Whether to include arguments in the log message.

True
log_result bool

Whether to include the return value in the log message.

True

Returns:

Type Description
Callable[[Callable[P, R]], Callable[P, R]]

A decorator that applies the logging annotation to a method.

Source code in plugins/spakky-logging/src/spakky/plugins/logging/annotation.py
def logged(
    enable_masking: bool = True,
    masking_keys: list[str] | None = None,
    slow_threshold_ms: float | None = None,
    max_result_length: int | None = None,
    log_args: bool = True,
    log_result: bool = True,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Decorator for enabling automatic method logging.

    Args:
        enable_masking: Whether to mask sensitive data in logs.
        masking_keys: List of keys whose values should be masked.
            When empty, the global :attr:`LoggingConfig.mask_keys` is used.
        slow_threshold_ms: Per-method slow-call warning threshold.
            ``None`` falls back to :attr:`LoggingConfig.slow_threshold_ms`.
        max_result_length: Maximum repr length for the return value.
            ``None`` falls back to :attr:`LoggingConfig.max_result_length`.
        log_args: Whether to include arguments in the log message.
        log_result: Whether to include the return value in the log message.

    Returns:
        A decorator that applies the logging annotation to a method.
    """

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        return Logged(
            enable_masking=enable_masking,
            masking_keys=masking_keys or [],
            slow_threshold_ms=slow_threshold_ms,
            max_result_length=max_result_length,
            log_args=log_args,
            log_result=log_result,
        )(func)

    return decorator

options: show_root_heading: false

Configuration

spakky.plugins.logging.config

Logging configuration for Spakky Framework.

Provides a Configuration that controls how the framework's logging system is set up: format, levels, masking, and slow-call thresholds.

LogFormat

Bases: StrEnum

Supported log output formats.

LoggingConfig()

Bases: BaseSettings

Configuration for the Spakky logging system.

Attributes:

Name Type Description
level int

Root logger level (e.g. logging.INFO, logging.DEBUG).

format LogFormat

Output format — text, json, or pretty.

date_format str

strftime pattern for timestamps.

package_levels dict[str, int]

Per-logger level overrides, keyed by logger name.

mask_keys list[str]

Global list of sensitive keys to mask in @Logging output.

mask_replacement str

Replacement text for masked values.

slow_threshold_ms float

Millisecond threshold for slow-call warnings in @Logging.

max_result_length int

Maximum character length for result repr in @Logging.

Source code in plugins/spakky-logging/src/spakky/plugins/logging/config.py
def __init__(self) -> None:
    super().__init__()

options: show_root_heading: false

Context

spakky.plugins.logging.context

Log context management using contextvars.

Provides a thread-safe and async-safe mechanism for binding contextual key-value pairs to log records. Bound values are automatically available to all loggers within the same execution context (asyncio task or thread).

LogContext

Manages contextual key-value pairs for structured logging.

Values bound via :meth:bind are propagated through contextvars and injected into every log record by :class:ContextInjectingFilter.

bind(**kwargs) classmethod

Add key-value pairs to the current log context.

Parameters:

Name Type Description Default
**kwargs str

Key-value pairs to bind.

{}
Source code in plugins/spakky-logging/src/spakky/plugins/logging/context.py
@classmethod
def bind(cls, **kwargs: str) -> None:
    """Add key-value pairs to the current log context.

    Args:
        **kwargs: Key-value pairs to bind.
    """
    current = _log_context.get().copy()
    current.update(kwargs)
    _log_context.set(current)

unbind(*keys) classmethod

Remove keys from the current log context.

Parameters:

Name Type Description Default
*keys str

Keys to remove.

()
Source code in plugins/spakky-logging/src/spakky/plugins/logging/context.py
@classmethod
def unbind(cls, *keys: str) -> None:
    """Remove keys from the current log context.

    Args:
        *keys: Keys to remove.
    """
    current = _log_context.get().copy()
    for key in keys:
        current.pop(key, None)
    _log_context.set(current)

clear() classmethod

Remove all key-value pairs from the current log context.

Source code in plugins/spakky-logging/src/spakky/plugins/logging/context.py
@classmethod
def clear(cls) -> None:
    """Remove all key-value pairs from the current log context."""
    _log_context.set({})

get() classmethod

Return a copy of the current log context.

Returns:

Type Description
dict[str, str]

Current context key-value pairs.

Source code in plugins/spakky-logging/src/spakky/plugins/logging/context.py
@classmethod
def get(cls) -> dict[str, str]:
    """Return a copy of the current log context.

    Returns:
        Current context key-value pairs.
    """
    return _log_context.get().copy()

scope(**kwargs) classmethod

Temporarily bind values for the duration of a with block.

Previous context is restored when the block exits.

Parameters:

Name Type Description Default
**kwargs str

Key-value pairs to bind within the scope.

{}

Yields:

Type Description
Generator[None]

None

Source code in plugins/spakky-logging/src/spakky/plugins/logging/context.py
@classmethod
@contextmanager
def scope(cls, **kwargs: str) -> Generator[None]:
    """Temporarily bind values for the duration of a ``with`` block.

    Previous context is restored when the block exits.

    Args:
        **kwargs: Key-value pairs to bind within the scope.

    Yields:
        None
    """
    previous = _log_context.get().copy()
    merged = {**previous, **kwargs}
    _log_context.set(merged)
    try:
        yield
    finally:
        _log_context.set(previous)

options: show_root_heading: false

Filters

spakky.plugins.logging.filters

Logging filter that injects LogContext values into log records.

Attaches all key-value pairs from the current :class:LogContext as attributes on every :class:logging.LogRecord, making them available to formatters.

ContextInjectingFilter

Bases: Filter

Filter that injects :class:LogContext values into log records.

When added to a logger or handler, every record will carry the current context as extra attributes and a consolidated context dict attribute.

filter(record)

Inject context values into the log record.

Parameters:

Name Type Description Default
record LogRecord

The log record to augment.

required

Returns:

Type Description
bool

Always True — this filter never suppresses records.

Source code in plugins/spakky-logging/src/spakky/plugins/logging/filters.py
def filter(self, record: logging.LogRecord) -> bool:
    """Inject context values into the log record.

    Args:
        record: The log record to augment.

    Returns:
        Always ``True`` — this filter never suppresses records.
    """
    context = LogContext.get()
    record.context = context  # type: ignore[attr-defined]
    for key, value in context.items():
        if not hasattr(record, key):
            setattr(record, key, value)
    return True

options: show_root_heading: false

Formatters

spakky.plugins.logging.formatters

Log formatters for Spakky Framework.

Provides three output formats: - Text: Traditional human-readable single-line logs. - JSON: Machine-parseable structured logs (one JSON object per line). - Pretty: Coloured multi-column layout for local development.

SpakkyTextFormatter(datefmt=DEFAULT_DATE_FORMAT)

Bases: Formatter

Human-readable single-line formatter.

Format::

2026-03-15T14:30:00+09:00 | INFO  | myapp.service | message

Initialize the text formatter.

Parameters:

Name Type Description Default
datefmt str

Date format string for timestamps.

DEFAULT_DATE_FORMAT
Source code in plugins/spakky-logging/src/spakky/plugins/logging/formatters.py
def __init__(self, datefmt: str = DEFAULT_DATE_FORMAT) -> None:
    """Initialize the text formatter.

    Args:
        datefmt: Date format string for timestamps.
    """
    super().__init__(datefmt=datefmt)

format(record)

Format the log record as a pipe-separated text line.

Parameters:

Name Type Description Default
record LogRecord

The log record to format.

required

Returns:

Type Description
str

Formatted log string.

Source code in plugins/spakky-logging/src/spakky/plugins/logging/formatters.py
def format(self, record: logging.LogRecord) -> str:
    """Format the log record as a pipe-separated text line.

    Args:
        record: The log record to format.

    Returns:
        Formatted log string.
    """
    timestamp = datetime.fromtimestamp(
        record.created,
        tz=timezone.utc,
    ).astimezone()
    ts_str = timestamp.strftime(self.datefmt or DEFAULT_DATE_FORMAT)

    parts = [
        ts_str,
        f"{record.levelname:<5}",
        record.name,
    ]

    context: dict[str, str] = getattr(record, "context", {})
    if context:
        ctx_str = " ".join(f"{k}={v}" for k, v in context.items())
        parts.append(ctx_str)

    parts.append(record.getMessage())

    line = self.SEPARATOR.join(parts)

    if record.exc_info and not record.exc_text:
        record.exc_text = self.formatException(record.exc_info)
    if record.exc_text:
        line = f"{line}\n{record.exc_text}"
    if record.stack_info:
        line = f"{line}\n{record.stack_info}"
    return line

SpakkyJsonFormatter(datefmt=DEFAULT_DATE_FORMAT)

Bases: Formatter

Structured JSON formatter (one object per line).

Output::

{"timestamp":"...","level":"INFO","logger":"...","message":"...","context_id":"..."}

Initialize the JSON formatter.

Parameters:

Name Type Description Default
datefmt str

Date format string for timestamps.

DEFAULT_DATE_FORMAT
Source code in plugins/spakky-logging/src/spakky/plugins/logging/formatters.py
def __init__(self, datefmt: str = DEFAULT_DATE_FORMAT) -> None:
    """Initialize the JSON formatter.

    Args:
        datefmt: Date format string for timestamps.
    """
    super().__init__(datefmt=datefmt)

format(record)

Format the log record as a JSON string.

Parameters:

Name Type Description Default
record LogRecord

The log record to format.

required

Returns:

Type Description
str

JSON-encoded log string.

Source code in plugins/spakky-logging/src/spakky/plugins/logging/formatters.py
def format(self, record: logging.LogRecord) -> str:
    """Format the log record as a JSON string.

    Args:
        record: The log record to format.

    Returns:
        JSON-encoded log string.
    """
    timestamp = datetime.fromtimestamp(
        record.created,
        tz=timezone.utc,
    ).astimezone()
    ts_str = timestamp.strftime(self.datefmt or DEFAULT_DATE_FORMAT)

    entry: dict[str, object] = {
        "timestamp": ts_str,
        "level": record.levelname,
        "logger": record.name,
        "message": record.getMessage(),
    }

    context: dict[str, str] = getattr(record, "context", {})
    if context:
        entry.update(context)

    if record.exc_info:
        entry["exception"] = self.formatException(record.exc_info)
    if record.stack_info:
        entry["stack_info"] = record.stack_info

    return json.dumps(entry, default=str, ensure_ascii=False)

SpakkyPrettyFormatter

Bases: Formatter

Coloured multi-column formatter for local development.

Format::

14:30:00.123 | INFO  | myapp.service | req-abc123 |
message text here

format(record)

Format the log record with colours and alignment.

Parameters:

Name Type Description Default
record LogRecord

The log record to format.

required

Returns:

Type Description
str

Colourised, aligned log string.

Source code in plugins/spakky-logging/src/spakky/plugins/logging/formatters.py
def format(self, record: logging.LogRecord) -> str:
    """Format the log record with colours and alignment.

    Args:
        record: The log record to format.

    Returns:
        Colourised, aligned log string.
    """
    timestamp = datetime.fromtimestamp(
        record.created,
        tz=timezone.utc,
    ).astimezone()
    ts_str = timestamp.strftime(PRETTY_TIME_FORMAT) + f"{record.msecs:03.0f}"

    color = self.LEVEL_COLORS.get(record.levelno, "")
    reset = self.RESET
    dim = self.DIM

    context: dict[str, str] = getattr(record, "context", {})
    ctx_str = " ".join(f"{k}={v}" for k, v in context.items()) if context else ""

    header = (
        f"{dim}{ts_str}{reset}"
        f" {dim}|{reset}"
        f" {color}{record.levelname:<5}{reset}"
        f" {dim}|{reset}"
        f" {record.name}"
    )
    if ctx_str:
        header = f"{header} {dim}|{reset} {ctx_str}"

    message = record.getMessage()
    line = f"{header} {dim}|{reset}\n  {message}"

    if record.exc_info and not record.exc_text:
        record.exc_text = self.formatException(record.exc_info)
    if record.exc_text:
        line = f"{line}\n{record.exc_text}"
    if record.stack_info:
        line = f"{line}\n{record.stack_info}"
    return line

options: show_root_heading: false

Post Processor

spakky.plugins.logging.post_processor

Post-processor that configures Python logging on application start.

Reads :class:LoggingConfig from the container and applies:

  • Root logger level and handler with the selected formatter
  • Per-package level overrides
  • :class:ContextInjectingFilter on the root logger

LoggingSetupPostProcessor()

Bases: IPostProcessor, IContainerAware

Post-processor that configures Python logging from LoggingConfig.

Runs once on the first post_process call. Subsequent calls are no-ops for the setup logic; the pod itself is returned unchanged.

Source code in plugins/spakky-logging/src/spakky/plugins/logging/post_processor.py
def __init__(self) -> None:
    super().__init__()
    self.__configured = False

post_process(pod)

Configure logging on first invocation, then pass through.

Parameters:

Name Type Description Default
pod object

The Pod instance being processed.

required

Returns:

Type Description
object

The unmodified Pod instance.

Source code in plugins/spakky-logging/src/spakky/plugins/logging/post_processor.py
def post_process(self, pod: object) -> object:
    """Configure logging on first invocation, then pass through.

    Args:
        pod: The Pod instance being processed.

    Returns:
        The unmodified Pod instance.
    """
    if not self.__configured:
        self.__configured = True
        self._configure()
    return pod

options: show_root_heading: false