Logging Service

A logging service is a compact way to show the Strategy pattern. Three things vary independently: which messages get through (level), how they look (format), and where they go (destination). Keep those axes separate and every extension becomes a new class.

Requirements#

The logger has to filter, format and route each message:

1. A client logs one message at a time.
2. Each message has a level: DEBUG < INFO < WARN < ERROR. A message is processed only if its level is at or above the configured minimum.
3. Messages can be formatted as plain text, JSON or XML.
4. Messages can go to one or more destinations: console, file, and so on.

Actor: the client code calling the logger.

Out of scope: UI, concurrency, batching.

Core entities#

The two interfaces are the extension points:

Relationships#

The logger only ever talks to interfaces:

• Logger → Formatter (one) and Destination (many): both injected, so the logger depends only on the interfaces.
• Logger uses Message per call.

Key flows#

There is one main flow.

log(message):

1. If message.level < min_level, skip it.
2. Format once with the formatter.
3. Send the formatted text to every destination; collect failures.
4. Report success, partial failure, or skipped, so the caller can tell them apart.

Code#

One formatter and destination of each kind is enough to show the idea:

import json
from abc import ABC, abstractmethod
from datetime import datetime
from enum import Enum
from typing import List, Tuple


class LogLevel(Enum):
    DEBUG = 1
    INFO = 2
    WARN = 3
    ERROR = 4


class Message:
    def __init__(self, text: str, level: LogLevel,
                 timestamp: datetime | None = None):
        self.text = text
        self.level = level
        self.timestamp = timestamp or datetime.now()


class Formatter(ABC):
    @abstractmethod
    def format(self, message: Message) -> str:
        pass


class Destination(ABC):
    @abstractmethod
    def send(self, text: str) -> bool:
        pass


class PlainFormatter(Formatter):
    def format(self, message: Message) -> str:
        time = message.timestamp.isoformat()
        return f"[{message.level.name}] {time} - {message.text}"


class JsonFormatter(Formatter):
    def format(self, message: Message) -> str:
        return json.dumps({
            "level": message.level.name,
            "time": message.timestamp.isoformat(),
            "message": message.text,
        })


class ConsoleDestination(Destination):
    def send(self, text: str) -> bool:
        print(text)
        return True


class FileDestination(Destination):
    def __init__(self, path: str):
        self.path = path

    def send(self, text: str) -> bool:
        try:
            with open(self.path, "a") as f:
                f.write(text + "\n")
            return True
        except OSError:
            return False  # report failure instead of crashing the caller


class Logger:
    def __init__(self, min_level: LogLevel, formatter: Formatter,
                 destinations: List[Destination]):
        self.min_level = min_level
        self.formatter = formatter
        self.destinations = destinations

    def log(self, message: Message) -> Tuple[str, bool]:
        if message.level.value < self.min_level.value:
            return "Ignored: below min level", False

        text = self.formatter.format(message)  # format once, send many
        failed = [d for d in self.destinations if not d.send(text)]
        if not failed:
            return "All destinations succeeded", True
        return f"{len(failed)} destination(s) failed", False

    # convenience helpers
    def info(self, text: str):
        return self.log(Message(text, LogLevel.INFO))

    def error(self, text: str):
        return self.log(Message(text, LogLevel.ERROR))


if __name__ == "__main__":
    console = [ConsoleDestination()]
    logger = Logger(LogLevel.INFO, PlainFormatter(), console)
    print(logger.info("server started"))                         # printed
    print(logger.log(Message("verbose trace", LogLevel.DEBUG)))  # ignored

    json_logger = Logger(LogLevel.WARN, JsonFormatter(), console)
    json_logger.error("disk almost full")

Extensions and trade-offs#

Because each axis is its own interface, most follow-ups are additive:

• New format or destination: add a class; Logger doesn't change (Open/Closed).
• Per-destination levels or formats: wrap each destination with its own level and formatter (a small "handler" object), like Python's logging module does.
• Async logging: put messages on a queue and let a background worker write them, so slow destinations don't block callers. Trade-off: messages can be lost on crash.
• Singleton logger: a global logger is convenient, but injecting it keeps code testable.
• Concurrency: if multiple threads log to one file, guard writes with a lock or use a single writer thread.
• Chain of Responsibility is another common framing: each handler decides whether to process the message and passes it on.

Key takeaways#

• Separate the three axes: level filtering, formatting, destination.
• Formatters and destinations are strategies injected into the logger.
• Format once, fan out to many destinations, and report partial failures.
• Async queues and per-destination handlers are the usual follow-ups.