Source code for pas.importer.output_handlers

from __future__ import annotations

from datetime import datetime
from typing import Any, TYPE_CHECKING

if TYPE_CHECKING:
    import logging



[docs]
class DatabaseOutputHandler:
    """Collects import messages for database storage in ImportLog.

    Facilitates dataflow by capturing detailed logging separately.
    """

    def __init__(self) -> None:

[docs]
        self.messages: list[dict[str, Any]] = []



[docs]
    def info(self, message: str) -> None:
        self._add_message('info', message)



[docs]
    def success(self, message: str) -> None:
        self._add_message('success', message)



[docs]
    def error(self, message: str) -> None:
        self._add_message('error', message)



[docs]
    def _add_message(self, level: str, message: str) -> None:
        """Add a message with timestamp and level to the collection."""
        self.messages.append({
            'timestamp': datetime.utcnow().isoformat(),
            'level': level,
            'message': message
        })



[docs]
    def get_messages(self) -> list[dict[str, Any]]:
        """Return all collected messages."""
        return self.messages.copy()



[docs]
    def clear_messages(self) -> None:
        """Clear all collected messages."""
        self.messages.clear()





[docs]
class LogOutputHandler:
    """Logs import messages using Python logging."""

    def __init__(self, logger: logging.Logger) -> None:

[docs]
        self.logger = logger



[docs]
    def info(self, message: str) -> None:
        self.logger.info(message)



[docs]
    def success(self, message: str) -> None:
        self.logger.info(f'SUCCESS: {message}')



[docs]
    def error(self, message: str) -> None:
        self.logger.error(message)