from __future__ import annotations
import click
import logging
from typing import Protocol, Any
[docs]
log = logging.getLogger('onegov.org.cronjobs')
[docs]
class OutputHandler(Protocol):
"""Protocol for handling output during KUB import operations."""
[docs]
def info(self, message: str) -> None:
"""Output informational message."""
...
[docs]
def success(self, message: str) -> None:
"""Output success message."""
...
[docs]
def error(self, message: str) -> None:
"""Output error message."""
...
[docs]
class ClickOutputHandler:
"""OutputHandler implementation for CLI using click."""
[docs]
def info(self, message: str) -> None:
click.echo(message)
[docs]
def success(self, message: str) -> None:
click.echo(f'✓ {message}')
[docs]
def error(self, message: str) -> None:
click.echo(f'✗ {message}', err=True)
[docs]
class LogOutputHandler:
"""OutputHandler implementation for cronjobs using logging."""
[docs]
def info(self, message: str) -> None:
log.info(message)
[docs]
def success(self, message: str) -> None:
log.info(message)
[docs]
def error(self, message: str) -> None:
log.error(message)
[docs]
class CompositeOutputHandler:
"""OutputHandler that forwards messages to multiple handlers."""
def __init__(self, *handlers: OutputHandler):
[docs]
self.handlers = handlers
[docs]
def info(self, message: str) -> None:
for handler in self.handlers:
handler.info(message)
[docs]
def success(self, message: str) -> None:
for handler in self.handlers:
handler.success(message)
[docs]
def error(self, message: str) -> None:
for handler in self.handlers:
handler.error(message)
[docs]
def get_messages(self) -> list[dict[str, Any]]:
"""Get messages from the first handler that supports it."""
for handler in self.handlers:
if hasattr(handler, 'get_messages'):
return handler.get_messages()
return []