Logger

Below is a single self-contained Singleton logger class implementing:

Singleton behavior (Logger() always returns the same instance)
Pluggable handlers (console, file, SQLite, OpenSearch)
Configurable at initialization (e.g., via an env parameter or explicit handler list)
Basic log level shortcut methods (info, error, etc.)

You can use and extend this as a drop-in logger in your project.

import threading
import sqlite3
from datetime import datetime
import os
import sys

# Optional: opensearchpy dependency
try:
    from opensearchpy import OpenSearch
except ImportError:
    OpenSearch = None

class Logger:
    _instance = None
    _lock = threading.Lock()

    class _ConsoleHandler:
        def log(self, level, message, extra_data=None):
            print(f"[{level}] {datetime.utcnow().isoformat()} {message} {extra_data or ''}")

    class _FileHandler:
        def __init__(self, filename):
            self.filename = filename
        def log(self, level, message, extra_data=None):
            with open(self.filename, 'a') as f:
                line = f"[{level}] {datetime.utcnow().isoformat()} {message} {extra_data or ''}\\n"
                f.write(line)

    class _SQLiteHandler:
        def __init__(self, dbfile):
            self.conn = sqlite3.connect(dbfile, check_same_thread=False)
            self.conn.execute('''
                CREATE TABLE IF NOT EXISTS logs (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    ts TEXT,
                    level TEXT,
                    message TEXT,
                    extra TEXT
                )
            ''')
            self.conn.commit()
        def log(self, level, message, extra_data=None):
            ts = datetime.utcnow().isoformat()
            with self.conn:
                self.conn.execute('INSERT INTO logs (ts, level, message, extra) VALUES (?, ?, ?, ?)',
                    (ts, level, message, str(extra_data) if extra_data else '')
                )
        def close(self):
            self.conn.close()

    class _OpenSearchHandler:
        def __init__(self, host, port, index):
            if not OpenSearch:
                raise ImportError("opensearch-py is not installed")
            self.client = OpenSearch([{'host': host, 'port': port}])
            self.index = index
        def log(self, level, message, extra_data=None):
            doc = {
                "timestamp": datetime.utcnow().isoformat(),
                "level": level,
                "message": message,
                "extra": extra_data
            }
            self.client.index(index=self.index, body=doc)

    def __new__(cls, *args, **kwargs):
        with cls._lock:
            if not cls._instance:
                cls._instance = super().__new__(cls)
                cls._instance._initialized = False
        return cls._instance

    def __init__(self, env='dev', file_path='app.log', sqlite_path='app_logs.db',
                 os_host=None, os_port=9200, os_index='app_logs'):
        if self._initialized:
            return

        self.handlers = []
        # Configure handlers based on 'env'
        if env == 'dev':
            self.handlers.append(self._ConsoleHandler())
            self.handlers.append(self._FileHandler(file_path))
        elif env == 'prod':
            self.handlers.append(self._FileHandler(file_path))
            self.handlers.append(self._SQLiteHandler(sqlite_path))
            if os_host:
                self.handlers.append(self._OpenSearchHandler(os_host, os_port, os_index))
        else:
            self.handlers.append(self._ConsoleHandler())

        self._initialized = True

    def add_handler(self, handler):
        self.handlers.append(handler)

    def log(self, level, message, extra_data=None):
        for handler in self.handlers:
            try:
                handler.log(level, message, extra_data)
            except Exception as e:
                print(f"[Logger] Handler error: {e}", file=sys.stderr)

    def info(self, message, extra_data=None):
        self.log("INFO", message, extra_data)
    def error(self, message, extra_data=None):
        self.log("ERROR", message, extra_data)
    def debug(self, message, extra_data=None):
        self.log("DEBUG", message, extra_data)
    def warn(self, message, extra_data=None):
        self.log("WARN", message, extra_data)
    def close(self):
        for handler in self.handlers:
            if hasattr(handler, 'close'):
                handler.close()

# Example usage (can be removed/commented as needed)
if __name__ == '__main__':
    # Set env='prod' and fill os_host for OpenSearch; default is dev (console + file)
    logger = Logger(env='dev', file_path='demo.log', sqlite_path='test_logs.db')
    logger.info('App started')
    logger.error('Something went wrong', {'reason': 'testing', 'user': 'alice'})

    # Always singleton
    logger2 = Logger()
    assert logger is logger2

    # To use OpenSearch in production:
    # logger = Logger(env='prod', os_host='localhost', os_port=9200, os_index='myapp_logs')

Usage Summary¶

Logger() gives the singleton.
Select env 'dev' (console + file), 'prod' (file + sqlite + optional OpenSearch), or customize.
Add your own handlers via logger.add_handler(custom_handler).
Logging: logger.info('msg'), logger.error('error!'), etc.