scietex.logging

scietex.logging is an asynchronous logging package designed for high-performance applications that require non-blocking logging. It uses asyncio to manage log message queues and provides multiple backends, such as console, file, Redis, Valkey, and MQTT logging, allowing for easy extension to other logging targets.

Built on the standard logging module. scietex.logging is not a replacement for Python's standard logging — it extends it. Every handler subclasses logging.Handler (through AsyncLoggingHandler), so handlers attach to ordinary loggers with logger.addHandler(handler) and receive records through the normal logging pipeline (logger.info(...) → emit()). Formatters subclass logging.Formatter. Standard levels, logger.setLevel(), logger.addHandler(), logger.removeHandler(), and logging.shutdown() all work unchanged, and you can mix scietex.logging handlers with standard-library handlers on the same logger. The only difference is that scietex.logging handlers process records asynchronously instead of synchronously in the calling thread.

Features

• Asynchronous Logging: Log messages are queued and handled asynchronously, reducing impact on application performance.
• Loop-Independent emit: emit() is thread-safe and may be called from any thread — including one with no running asyncio loop — so you can log from worker threads, thread pools, and callbacks without dropping records.
• Multiple Backends: Supports console, file, Redis, Valkey, and MQTT logging out of the box.
• Themes and Color: Opt-in ANSI color for console output via the Palette/LoggingTheme theme API — built-in MONOCHROME, SCIETEX_LIGHT, and SCIETEX_DARK themes plus resolve_color TTY auto-detection, applied through theme=/color= on ScietexFormatter and ConsoleHandler.
• Flexible Logging Levels: Compatible with Python's standard logging levels (DEBUG, INFO, WARNING, ERROR, CRITICAL).
• Optional Dependencies: Only installs dependencies for the specific backends you need.

Examples

Explore the examples/ directory to see usage examples that demonstrate how to set up and work with scietex.logging. Each example provides a practical setup for different logging scenarios, including basic console logging and Redis-based logging.

For detailed descriptions of each example, refer to the Examples README.

Requirements

• Python 3.10+
• Additional dependencies for specific backends:
  • Redis support: redis (pip install scietex.logging[redis])
  • Valkey support: valkey-glide (pip install scietex.logging[valkey])
  • MQTT support: aiomqtt (pip install scietex.logging[mqtt])

Installation

Install the base package with:

pip install scietex.logging

To install all optional dependencies (including Redis, Valkey, and MQTT support), use:

pip install scietex.logging[all]

Or, to install individual dependencies as needed:

pip install scietex.logging[redis]   # For Redis logging
pip install scietex.logging[valkey]  # For Valkey logging
pip install scietex.logging[mqtt]    # For MQTT logging

Basic Usage

Console Logging

The following example shows how to set up asynchronous console logging.

import logging
from scietex.logging import ConsoleHandler
import asyncio

# Set up logger and handler
logger = logging.getLogger("MyAsyncLogger")
logger.setLevel(logging.DEBUG)
handler = ConsoleHandler()
logger.addHandler(handler)


async def main():
    await handler.start_logging()
    logger.info("This is an asynchronous log message")
    await handler.stop_logging()


asyncio.run(main())

Redis Logging

This example demonstrates logging to a Redis stream.

import logging
from scietex.logging import AsyncRedisHandler
import asyncio

# Set up logger and Redis handler
logger = logging.getLogger("MyAsyncLogger")
logger.setLevel(logging.DEBUG)
handler = AsyncRedisHandler(stream_name="my_log_stream")
logger.addHandler(handler)


async def main():
    await handler.start_logging()
    logger.error("This error message will be logged to Redis!")
    await handler.stop_logging()


asyncio.run(main())

Valkey Logging

This example demonstrates logging to a Valkey stream.

import logging
from scietex.logging import AsyncValkeyHandler
import asyncio

# Set up logger and Valkey handler
logger = logging.getLogger("MyAsyncLogger")
logger.setLevel(logging.DEBUG)
handler = AsyncValkeyHandler(stream_name="my_log_stream", stream_maxlen=1000)
logger.addHandler(handler)


async def main():
    await handler.start_logging()
    logger.error("This error message will be logged to Valkey!")
    await handler.stop_logging()


asyncio.run(main())

MQTT Logging

This example demonstrates publishing log records to an MQTT topic.

import logging
from scietex.logging import AsyncMqttHandler
import asyncio

# Set up logger and MQTT handler
logger = logging.getLogger("MyAsyncLogger")
logger.setLevel(logging.DEBUG)
handler = AsyncMqttHandler(topic="my/log/topic", message_expiry=3600)
logger.addHandler(handler)


async def main():
    await handler.start_logging()
    logger.error("This error message will be logged to MQTT!")
    await handler.stop_logging()


asyncio.run(main())

File Logging

This example demonstrates logging to a file, including structured JSON output.

import logging
from scietex.logging import AsyncFileHandler, JsonFormatter
import asyncio

# Set up logger and file handler
logger = logging.getLogger("MyAsyncLogger")
logger.setLevel(logging.DEBUG)
handler = AsyncFileHandler("app.log")
logger.addHandler(handler)

# JSON output (optional): swap the default formatter for JsonFormatter
# handler.setFormatter(JsonFormatter())


async def main():
    await handler.start_logging()
    logger.error("This error message will be logged to a file!")
    await handler.stop_logging()


asyncio.run(main())

File logging needs no extra dependency. Rotation variants (AsyncRotatingFileHandler, AsyncTimedRotatingFileHandler, AsyncWatchedFileHandler) mirror the stdlib classes of the same name.

Themes and Color

Console output is monochrome by default. Pass a theme to ConsoleHandler to colorize it, or apply a theme directly to a ScietexFormatter:

import logging
from scietex.logging import (
    SCIETEX_DARK,
    SCIETEX_LIGHT,
    ConsoleHandler,
    ScietexFormatter,
)

# Theme on the console handler — color is auto-detected from a TTY stdout.
handler = ConsoleHandler(theme=SCIETEX_DARK)

# Theme on the formatter — `color=True` forces color explicitly.
formatter = ScietexFormatter(theme=SCIETEX_LIGHT, color=True)
handler.setFormatter(formatter)

Color is opt-in and emits 24-bit truecolor ANSI sequences. resolve_color enables it only for a TTY sys.stdout (or on an explicit FORCE_COLOR / force_color request), so redirected and file output stay plain.

Configuration

scietex.logging is designed to allow easy configuration of additional backends and custom logging formats:

Formatting: Use Python’s standard logging Formatter to customize output. For example, to log timestamps in ISO format:

formatter = logging.Formatter(
    fmt="%(asctime)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%dT%H:%M:%SZ"
)
handler.setFormatter(formatter)

A formatter affects the console (stdout) and file output only; broker handlers no longer accept a formatter= keyword and build their payloads from the log record directly — its name field is the record's logger name (record.name) — so they are invariant under setFormatter. See docs/configuration.md.

Extending scietex.logging

To add support for additional logging backends, subclass AsyncBrokerHandler and implement connect(), disconnect(), and send_message() methods. AsyncLoggingHandler is the pure-machinery base that owns the queue/worker infrastructure but no sink of its own. ConsoleHandler and AsyncFileHandler (and its rotation variants) are the concrete handlers that each subclass AsyncLoggingHandler directly and register their own backend — the console and file sinks respectively. AsyncBrokerHandler is the abstract base for the message-broker backends (Redis, Valkey, MQTT, and custom backends); it also subclasses AsyncLoggingHandler directly but declares connect(), disconnect(), and send_message() as abstract methods, so a broker backend subclasses AsyncBrokerHandler rather than the base. A handler emits only to the backend it registers, so console output requires adding a ConsoleHandler to the logger explicitly.

Example: Custom Database Handler

from scietex.logging import AsyncBrokerHandler


class AsyncPostgresHandler(AsyncBrokerHandler):
    def __init__(self, db_url):
        super().__init__(queue_name="postgres")
        self.db_url = db_url
        self._db_conn = None

    async def connect(self):
        import asyncpg

        self._db_conn = await asyncpg.connect(self.db_url)

    async def disconnect(self):
        if self._db_conn:
            await self._db_conn.close()

    async def send_message(self, record):
        await self._db_conn.execute(
            "INSERT INTO logs (level, message) VALUES ($1, $2)",
            record["level"],
            record["message"],
        )

Contributing

Contributions are welcome! If you find a bug or want to add a feature, please open an issue or submit a pull request.

License

This project is licensed under the MIT License.