Logging & Observability

ZodiacCore provides a pre-configured, production-ready logging system based on Loguru. It is designed for observability, supporting JSON structured logs and automatic Trace ID correlation.

1. Core Concepts

Structured Logging

By default, ZodiacCore outputs logs in JSON format. This is ideal for production environments (e.g., ELK stack, Datadog, CloudWatch) as it makes log parsing and searching significantly easier.

Trace ID Correlation

Every log message automatically includes a request_id if it was set during an active HTTP request or WebSocket connection (by TraceIDMiddleware). This allows you to correlate multiple log lines across different services for a single transaction.

---

2. Quick Setup

The most common way to initialize logging is in your application's entry point (main.py).

from zodiac_core.logging import setup_loguru

setup_loguru(
    level="INFO",
    json_format=True,        # Use JSON for production
    service_name="payment-service"
)

---

3. Advanced Configuration

Console & File Output

You can log to both the console and a file simultaneously.

from zodiac_core.logging import setup_loguru, LogFileOptions

setup_loguru(
    level="DEBUG",
    json_format=False,       # Use human-readable text for local dev
    log_file="logs/app.log",
    file_options=LogFileOptions(
        rotation="500 MB",
        retention="10 days",
        compression="zip"
    )
)

Passing Extra Sink Options

The console_options argument allows you to pass arbitrary keyword arguments directly to the Loguru add() method.

setup_loguru(
    console_options={"enqueue": True, "backtrace": True, "diagnose": True}
)

---

4. How to Log

Since ZodiacCore configures the standard loguru.logger, you can simply import and use it anywhere in your code.

from loguru import logger

def process_data(data):
    logger.info("Processing data", extra={"data_id": data.id})
    # If this runs during a request, 'request_id' is automatically added!

---

5. JSON Log Structure

A typical JSON log entry produced by ZodiacCore looks like this:

{
  "text": "2026-01-31 17:26:24.208 | INFO     | demo_r:read_item:23 - request: item_id=1\n",
  "record": {
    "elapsed": {
      "repr": "0:00:14.429585",
      "seconds": 14.429585
    },
    "exception": null,
    "extra": {
      "request_id": "98277dc9-27ca-4849-98f0-6097c3b41867",
      "service": "service"
    },
    "file": {
      "name": "demo_r.py",
      "path": "/Users/legolas/workspace/ZodiacCore-Py/demo_r.py"
    },
    "function": "read_item",
    "level": {
      "icon": "ℹ️",
      "name": "INFO",
      "no": 20
    },
    "line": 23,
    "message": "request: item_id=1",
    "module": "demo_r",
    "name": "demo_r",
    "process": {
      "id": 92473,
      "name": "MainProcess"
    },
    "thread": {
      "id": 140704462000000,
      "name": "MainThread"
    },
    "time": {
      "repr": "2026-01-31 17:26:24.208560+08:00",
      "timestamp": 1769851584.20856
    }
  }
}

---

6. API Reference

Logging Utilities

LogFileOptions

Bases: BaseModel

Configuration options for file logging.

Allows arbitrary extra arguments to be passed to loguru.add() via extra="allow".

Source code in zodiac_core/logging.py

class LogFileOptions(BaseModel):
    """
    Configuration options for file logging.

    Allows arbitrary extra arguments to be passed to loguru.add() via extra="allow".
    """

    rotation: str = "10 MB"
    retention: str = "1 week"
    compression: str = "zip"
    enqueue: bool = False
    encoding: str = "utf-8"

    model_config = ConfigDict(extra="allow")

setup_loguru(level='INFO', json_format=True, service_name='service', log_file=None, console_options=None, file_options=None)

Configure Loguru with automatic Trace ID injection and multi-destination output.

Parameters:

Name	Type	Description	Default
level	str	Logging level (INFO, DEBUG, etc.)	'INFO'
json_format	bool	Whether to output JSON (True) or Text (False). When True, the serialized JSON has an empty "text" field to avoid duplicating the message (see record.message); pass a custom "format" in console_options/file_options if you need a non-empty "text".	True
service_name	str	Name of the service (added to JSON logs).	'service'
log_file	Optional[str]	Optional file path to save logs.	None
console_options	Optional[Dict[str, Any]]	Extra kwargs to pass to the console sink (e.g. {"enqueue": True}).	None
file_options	Optional[LogFileOptions]	Configuration model (LogFileOptions) for file sink.	None

Source code in zodiac_core/logging.py

def setup_loguru(
    level: str = "INFO",
    json_format: bool = True,
    service_name: str = "service",
    log_file: Optional[str] = None,
    console_options: Optional[Dict[str, Any]] = None,
    file_options: Optional[LogFileOptions] = None,
):
    """
    Configure Loguru with automatic Trace ID injection and multi-destination output.

    Args:
        level: Logging level (INFO, DEBUG, etc.)
        json_format: Whether to output JSON (True) or Text (False). When True, the
            serialized JSON has an empty "text" field to avoid duplicating the message
            (see record.message); pass a custom "format" in console_options/file_options
            if you need a non-empty "text".
        service_name: Name of the service (added to JSON logs).
        log_file: Optional file path to save logs.
        console_options: Extra kwargs to pass to the console sink (e.g. {"enqueue": True}).
        file_options: Configuration model (LogFileOptions) for file sink.
    """
    # 1. Remove default handlers
    logger.remove()

    service = service_name

    # 2. Configure Patcher (Trace ID injection)
    def patcher(record):
        request_id = get_request_id()
        if request_id:
            record["extra"]["request_id"] = request_id
        record["extra"]["service"] = service

    logger.configure(patcher=patcher)

    # 3. Define Formatters
    def _dev_formatter(record):
        if "request_id" not in record["extra"]:
            record["extra"]["request_id"] = "-"
        return (
            "<green>{time:YYYYMMDD HH:mm:ss}</green> "
            "| {extra[service]} "
            "| {extra[request_id]} "
            "| {process.name} "
            "| {thread.name} "
            "| <cyan>{module}</cyan>.<cyan>{function}</cyan> "
            "| <level>{level}</level>: "
            "<level>{message}</level> "
            "| {file.path}:{line}\n"
        )

    # 4. Sink defaults shared by console and file (level, enqueue, format/serialize)
    # Empty format avoids duplicating message in "text" and "record.message" (see loguru#594)
    if json_format:
        _format_defaults: Dict[str, Any] = {"serialize": True, "format": lambda _: ""}
    else:
        _format_defaults = {"format": _dev_formatter}
    # enqueue=True for thread-safe sink
    _sink_defaults: Dict[str, Any] = {"level": level, "enqueue": True, **_format_defaults}

    def _apply_sink_defaults(config: Dict[str, Any], sink: Any) -> None:
        for key, value in _sink_defaults.items():
            config.setdefault(key, value)
        config.setdefault("sink", sink)

    # 5. Console sink
    c_config = console_options or {}
    _apply_sink_defaults(c_config, sys.stderr)
    logger.add(**c_config)

    # 6. File sink (if enabled)
    if log_file:
        if file_options is None:
            file_options = LogFileOptions()
        f_config = file_options.model_dump()
        _apply_sink_defaults(f_config, log_file)
        logger.add(**f_config)