如何将Python日志输出至Cosmos DB？能否自定义其日志处理器？

Great question! You absolutely can build a custom Cosmos DB log handler similar to FileHandler and use dictConfig to manage log levels—let's walk through how to implement this properly.

First, you'll need to create a custom handler by subclassing Python's built-in logging.Handler. The key method to override is emit(), which handles writing each log record to Cosmos DB.

Assuming you're using Azure Cosmos DB (the most common implementation), install the required package first:

pip install azure-cosmos

Here's the custom handler code with robust error handling:

import logging
from azure.cosmos import CosmosClient, exceptions
from datetime import datetime

class CosmosDBHandler(logging.Handler):
    def __init__(self, endpoint, key, database_name, container_name):
        super().__init__()
        # Initialize Cosmos DB client and container
        self.client = CosmosClient(endpoint, key)
        self.database = self.client.get_database_client(database_name)
        self.container = self.database.get_container_client(container_name)

    def emit(self, record):
        try:
            # Format log record into a structured JSON document for Cosmos DB
            log_entry = {
                "id": str(datetime.utcnow().timestamp()).replace(".", ""),  # Unique ID for Cosmos
                "logger_name": record.name,
                "log_level": record.levelname,
                "message": self.format(record),
                "timestamp": datetime.utcnow().isoformat(),
                "module": record.module,
                "function": record.funcName,
                "line_number": record.lineno
            }
            # Insert the log entry into Cosmos DB
            self.container.create_item(body=log_entry)
        except exceptions.CosmosHttpResponseError as e:
            # Gracefully handle Cosmos-specific errors so logging failures don't crash your app
            print(f"Failed to write log to Cosmos DB: {str(e)}")
        except Exception as e:
            print(f"Unexpected error in CosmosDBHandler: {str(e)}")

Now you can integrate this custom handler into your logging setup using dictConfig, just like you would with standard handlers. This lets you centrally control log levels, formatters, and handler behavior.

Here's a complete configuration example:

import logging.config

# Store these securely in production (use env vars or secrets managers, not hardcode!)
COSMOS_ENDPOINT = "your-cosmos-endpoint"
COSMOS_KEY = "your-cosmos-key"
COSMOS_DB_NAME = "logs-db"
COSMOS_CONTAINER_NAME = "app-logs"

logging_config = {
    "version": 1,
    "disable_existing_loggers": False,
    "formatters": {
        "detailed": {
            "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
        }
    },
    "handlers": {
        "cosmos": {
            "class": "__main__.CosmosDBHandler",  # Replace with your actual module path
            "level": "DEBUG",  # Minimum log level this handler will process
            "formatter": "detailed",
            "endpoint": COSMOS_ENDPOINT,
            "key": COSMOS_KEY,
            "database_name": COSMOS_DB_NAME,
            "container_name": COSMOS_CONTAINER_NAME
        },
        "console": {
            "class": "logging.StreamHandler",
            "level": "INFO",
            "formatter": "detailed"
        }
    },
    "loggers": {
        "my_app": {
            "handlers": ["cosmos", "console"],
            "level": "DEBUG",  # Root level for this logger (filters logs before they reach handlers)
            "propagate": False
        }
    }
}

# Apply the configuration
logging.config.dictConfig(logging_config)

# Test the setup
logger = logging.getLogger("my_app")
logger.debug("This debug message will only go to Cosmos DB")
logger.info("This info message shows up in both Cosmos and console")
logger.error("Error details are captured in structured format in Cosmos")

• Secure Credentials: Never hardcode Cosmos DB keys—use environment variables (e.g., os.getenv("COSMOS_KEY")) or a secrets manager like Azure Key Vault.
• Structured Logs: Storing logs as JSON in Cosmos DB makes it easy to query and filter (e.g., fetch all error logs from a specific module in the last hour).
• Level Control: You can set different levels for the logger and handler—for example, the logger could only pass INFO+ logs, while the handler captures everything it receives.

To directly answer your questions:

1. To send logs to Cosmos DB, create a custom handler like the one above and register it in your logging configuration.
2. Yes, you can build a Cosmos DB handler just like FileHandler, and dictConfig works perfectly to control log levels, formatters, and handler assignments.

内容的提问来源于stack exchange，提问作者Anirban Nag 'tintinmj'