Skip to main content

Platform-agnostic logging client for Artissist

Project description

Mosaic Logger - Python Client

Platform-agnostic logging client for the Mosaic Artist Assistant platform.

Features

  • 🎯 Pre-defined Events: 25 built-in event types with emoji support
  • 🎨 Emoji Integration: Configurable emoji prefixes for visual log scanning
  • 📊 Structured Logging: Rich metadata, metrics, and error information
  • 🔄 Context Management: Distributed tracing with correlation IDs
  • 🔌 Adapter Pattern: Multiple output destinations (console, file, cloud)
  • 🏭 Factory Pattern: Specialized loggers for different components
  • Async Support: Non-blocking logging with async/await
  • 🛡️ Type Safety: Full type hints and runtime validation

Installation

pip install mosaic-logger

Development Installation

pip install -e .[dev]

Quick Start

from mosaic_logger import LoggerFactory, LogEvent

# Create a backend logger
logger = LoggerFactory.create_backend_logger(
    service="my-service",
    environment="development", 
    emojis=True  # Enable emojis for development
)

# Basic logging
await logger.info("Service starting up", event=LogEvent.SYSTEM_START)
await logger.error("Database connection failed", event=LogEvent.ERROR_OCCURRED)

# With structured data
await logger.info(
    "User authenticated successfully",
    event=LogEvent.USER_AUTH,
    metadata={"user_id": "user_123", "method": "oauth"},
    metrics={"auth_duration_ms": 150}
)

Logger Types

Backend Logger

logger = LoggerFactory.create_backend_logger(
    service="api-server",
    environment="production",
    emojis=False  # Disabled for production
)

Agent Logger

logger = LoggerFactory.create_agent_logger(
    agent_id="conv_001",
    agent_type="conversation",
    environment="development",
    emojis=True
)

Infrastructure Logger

logger = LoggerFactory.create_infrastructure_logger(
    component="deployment-manager", 
    environment="production"
)

Context Management

Global Context

from mosaic_logger import ContextManager

# Set global context
ContextManager.set_context(LoggerContext(
    correlation_id="req_123",
    user_id="user_456"
))

# All subsequent logs will include this context
await logger.info("Processing request")  # Includes correlation_id and user_id

Scoped Context

# Temporary context for a specific operation
with ContextManager.context(user_id="user_789", operation="data_export"):
    await logger.info("Starting export")  # Includes user_id and operation
    await logger.info("Export completed")

Logger Context

# Create logger with permanent context
contextual_logger = logger.with_context(
    service_version="1.2.0",
    deployment_id="deploy_abc"
)

await contextual_logger.info("Service initialized")  # Includes service_version and deployment_id

Events and Emojis

Pre-defined Events

# System events
LogEvent.SYSTEM_START      # 🚀 System startup
LogEvent.ERROR_OCCURRED    # 🐛 Errors and exceptions
LogEvent.WARNING_ISSUED    # ⚠️  Warning conditions

# Business events  
LogEvent.USER_AUTH         # 👤 User authentication
LogEvent.PROJECT_LIFECYCLE # 📁 Project management
LogEvent.AI_INFERENCE      # 🧠 AI model operations

# Technical events
LogEvent.API_REQUEST       # 🔄 API calls
LogEvent.DATABASE_OPERATION # 💾 Database queries
LogEvent.PERFORMANCE_METRIC # ⚡ Performance measurements

Custom Events

from mosaic_logger import EmojiResolver, EmojiMapping

# Create custom emoji mappings
resolver = EmojiResolver()
resolver.add_custom_mapping("payment_processed", EmojiMapping(
    emoji="💳", 
    description="Payment processing events"
))

logger = LoggerFactory.create_logger(
    service="payment-service",
    environment="development", 
    adapters=["console"],
    emojis=True,
    emoji_resolver=resolver
)

# Use custom event
await logger.info("Payment completed", custom_event="payment_processed")
# Output: 💳 Payment completed

Error Handling

from mosaic_logger import ErrorInfo

try:
    # Some operation that might fail
    process_data()
except Exception as e:
    await logger.error(
        "Data processing failed",
        event=LogEvent.ERROR_OCCURRED,
        error=ErrorInfo(
            type=type(e).__name__,
            message=str(e),
            stack_trace=traceback.format_exc(),
            context={"batch_id": "batch_123"}
        )
    )

Metrics

from mosaic_logger import LogMetrics

# Performance metrics
await logger.info(
    "Database query completed",
    event=LogEvent.DATABASE_OPERATION,
    metrics=LogMetrics(
        duration_ms=45.2,
        count=150,
        bytes_processed=1024000
    )
)

# Business metrics
await logger.info(
    "User signup completed", 
    event=LogEvent.BUSINESS_METRIC,
    metadata={"user_type": "premium"},
    metrics=LogMetrics(
        count=1,
        custom_metrics={"revenue_impact": 29.99}
    )
)

Adapters

Console Adapter

# Development console output with colors
LoggerFactory.create_logger(
    service="my-service",
    environment="development",
    adapters=["console"],
    adapter_configs={
        "console": {"colors": True, "use_stderr": False}
    }
)

File Adapter

# Production file logging with rotation
LoggerFactory.create_logger(
    service="my-service", 
    environment="production",
    adapters=["file"],
    adapter_configs={
        "file": {
            "file_path": "logs/service.log",
            "format": "json",
            "rotate": True,
            "max_size_mb": 50,
            "max_files": 10
        }
    }
)

Multiple Adapters

# Both console and file output
LoggerFactory.create_backend_logger(
    service="api-service",
    environment="production",
    adapters=["console", "file"]
)

Synchronous Usage

For non-async contexts:

# Synchronous convenience methods (fire-and-forget)
logger.info_sync("Service started", event=LogEvent.SYSTEM_START)
logger.error_sync("Connection failed", event=LogEvent.ERROR_OCCURRED)

Configuration

Environment Variables

export MOSAIC_LOG_LEVEL=INFO
export MOSAIC_LOG_EMOJIS=true
export MOSAIC_LOG_FORMAT=json

Programmatic Configuration

config = {
    "service": "my-service",
    "environment": "production",
    "adapters": ["console", "file"],
    "emojis": False,
    "adapter_configs": {
        "console": {"colors": False},
        "file": {"file_path": "/var/log/my-service.log"}
    }
}

logger = LoggerFactory.create_logger(**config)

Development

Running Tests

pytest
pytest --cov=mosaic_logger  # With coverage

Code Formatting

black mosaic_logger/
mypy mosaic_logger/
flake8 mosaic_logger/

Build Package

python setup.py bdist_wheel

Integration Examples

See the examples/ directory for complete integration examples:

  • FastAPI backend service
  • Agent processing systems
  • Infrastructure deployment logging
  • Error handling patterns
  • Performance monitoring

License

MIT License - see LICENSE file for details.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distributions

No source distribution files available for this release.See tutorial on generating distribution archives.

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

artissist_logger-0.0.10-py3-none-any.whl (16.7 kB view details)

Uploaded Python 3

File details

Details for the file artissist_logger-0.0.10-py3-none-any.whl.

File metadata

File hashes

Hashes for artissist_logger-0.0.10-py3-none-any.whl
Algorithm Hash digest
SHA256 b0ef9cc4a60aeb98ac4c2b84cc3cdbcf56adbd0509e6427d18620b5e0db1ae86
MD5 3e6f138b65f2015bc6c2afc0ef236f41
BLAKE2b-256 dfed9d98339f804e61acedae46a289f50ae317bc451a9f03d32f1bc1016f4ac7

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page