character_forge_image/utils/logging_utils.md

logging_utils.py

Purpose

Centralized logging system for Nano Banana Streamlit. Provides both file-based logging (persistent) and memory-based logging (for UI display).

Responsibilities

• Set up loggers with consistent configuration
• Write logs to rotating file on disk
• Store recent logs in memory for UI display
• Provide utilities for structured logging
• Thread-safe log storage and retrieval
• Context managers for temporary log level changes

Dependencies

Imports

• logging - Python standard logging
• threading - Thread synchronization
• collections.deque - Thread-safe queue for log storage
• logging.handlers.RotatingFileHandler - Rotating log files
• config.settings.Settings - Log configuration

Used By

• All services - For logging generation progress
• All pages - For displaying logs in UI
• Backend clients - For logging API calls
• Utilities - For logging validation/file operations

Public Interface

Logger Setup

setup_logger(name: str, level: str = None) -> logging.Logger

Creates a fully configured logger with file and memory handlers.

Parameters:

• name: Logger name (usually module name or __name__)
• level: Optional log level override (default: from Settings)

Returns: Configured logger instance

Example:

from utils.logging_utils import setup_logger

logger = setup_logger('my_module')
logger.info("This logs to file and memory")

get_logger(name: str) -> logging.Logger

Get or create a logger (convenience function).

If logger doesn't exist, creates it with setup_logger(). If it exists, returns existing instance.

Example:

from utils.logging_utils import get_logger

logger = get_logger(__name__)
logger.info("Logging from this module")

Log Retrieval (for UI)

get_recent_logs(count: int = 100) -> List[str]

Returns recent log messages as a list.

Parameters:

• count: Maximum number of messages (default: 100)

Returns: List of formatted log strings

Example:

logs = get_recent_logs(50)
for log in logs:
    st.text(log)

get_recent_logs_as_string(count: int = 100) -> str

Returns recent log messages as a single string (one line per message).

Example:

logs_text = get_recent_logs_as_string(100)
st.text_area("Logs", logs_text)

clear_log_memory()

Clears the in-memory log queue.

Use when starting a new generation session.

get_log_count() -> int

Returns current number of messages in memory queue.

Utility Functions

log_function_call(logger, func_name: str, **kwargs)

Structured logging for function calls.

Example:

log_function_call(logger, "generate_image",
                  prompt="sunset", aspect_ratio="16:9")
# Logs: "Calling generate_image(prompt=sunset, aspect_ratio=16:9)"

log_stage(logger, stage: str, message: str)

Log a pipeline stage with separators.

Example:

log_stage(logger, "Stage 1/6", "Generating front portrait")
# Logs with ==== separators for visibility

log_error_with_context(logger, error: Exception, context: dict)

Log an error with additional context.

Example:

try:
    generate_image(...)
except Exception as e:
    log_error_with_context(logger, e, {
        'prompt': prompt,
        'backend': backend,
        'attempt': 3
    })

Context Managers

LoggingContext(logger_name: str, level: str)

Temporarily change log level.

Example:

from utils.logging_utils import LoggingContext

with LoggingContext('my_module', 'DEBUG'):
    # Temporarily log at DEBUG level
    logger.debug("This will be logged")
# Returns to original level

Internal Components

MemoryLogHandler

Custom logging handler that stores messages in thread-safe queue.

Inherits from logging.Handler and overrides emit() method.

_log_queue

Global deque(maxlen=1000) storing recent messages. Thread-safe with _log_lock.

_log_lock

Threading lock for synchronizing queue access.

Configuration

All configuration from Settings:

• LOG_LEVEL: Default log level ("INFO")
• LOG_FORMAT: Message format string
• LOG_DATE_FORMAT: Date format string
• LOG_FILE: Path to log file
• LOG_MAX_BYTES: Max file size before rotation (10MB)
• LOG_BACKUP_COUNT: Number of backup files (5)

Thread Safety

All log queue operations protected by _log_lock:

• _log_queue.append() - Adding messages
• list(_log_queue) - Reading messages
• _log_queue.clear() - Clearing queue

Safe to use from multiple threads (Streamlit reruns, background tasks).

Known Limitations

• Memory queue limited to 1000 messages (older messages discarded)
• File logs kept on disk (5 files × 10MB = 50MB max)
• No remote logging (syslog, cloud)
• No structured logging (JSON format)
• No log filtering by level in UI retrieval

Future Improvements

• Add structured logging (JSON Lines format)
• Add log level filtering for UI display
• Add log search/filtering capabilities
• Add remote logging option
• Add log analytics dashboard
• Add log export functionality
• Add colored console output for development

Usage Examples

Service Logging

from utils.logging_utils import get_logger

class CharacterForgeService:
    def __init__(self):
        self.logger = get_logger(__name__)

    def generate(self, prompt):
        self.logger.info(f"Starting generation: {prompt}")
        try:
            result = self._generate(prompt)
            self.logger.info("Generation successful")
            return result
        except Exception as e:
            self.logger.error(f"Generation failed: {e}")
            raise

UI Log Display

import streamlit as st
from utils.logging_utils import get_recent_logs_as_string

st.subheader("Generation Log")
logs = get_recent_logs_as_string(100)
st.text_area("Logs", logs, height=300)

if st.button("Clear Logs"):
    clear_log_memory()
    st.rerun()

Pipeline Stage Logging

from utils.logging_utils import get_logger, log_stage

logger = get_logger(__name__)

for stage_num in range(1, 7):
    log_stage(logger, f"Stage {stage_num}/6",
              f"Generating {view_name}")
    # ... generation code ...

Testing

• Test logger creation and configuration
• Test file handler writes to correct file
• Test memory handler stores messages
• Test log retrieval with various counts
• Test clear_log_memory()
• Test thread safety (concurrent access)
• Test context manager log level changes
• Test utility functions format correctly

Related Files

• config/settings.py - Log configuration constants
• All services - Use get_logger(name)
• All pages - Display logs with get_recent_logs()
• tests/test_logging_utils.py - Unit tests

Change History

• 2025-10-23: Initial creation for Streamlit migration
  • Extracted from character_forge.py (lines 44-143)
  • Generalized for all modules (not just Character Forge)
  • Added utility functions for structured logging
  • Added context manager for temporary log levels
  • Integrated with Settings configuration
  • Added comprehensive documentation