feat: add startup banner and enhanced event formatting for better visibility

adamrefaey · adamrefaey · commit 41bc6e5fa1d1 · 2025-12-30T11:53:57.000+02:00
diff --git a/src/asynctasq/core/worker.py b/src/asynctasq/core/worker.py
@@ -8,6 +8,9 @@
 from typing import Any
 import uuid
 
+from rich.panel import Panel
+from rich.table import Table
+
 from asynctasq.config import Config
 from asynctasq.drivers.base_driver import BaseDriver
 from asynctasq.drivers.retry_utils import calculate_retry_delay
@@ -73,6 +76,41 @@ def __init__(
         self._task_serializer = TaskSerializer(self.serializer)
         self._task_executor = TaskExecutor()
 
+    def _display_startup_banner(self) -> None:
+        """Display a beautiful startup banner with worker configuration."""
+        from rich.console import Console
+
+        console = Console()
+
+        # Create configuration table
+        config_table = Table.grid(padding=(0, 2))
+        config_table.add_column(style="cyan", justify="right")
+        config_table.add_column(style="bold white")
+
+        config_table.add_row("Worker ID", f"[bold blue]{self.worker_id}[/bold blue]")
+        config_table.add_row("Hostname", f"[dim]{self.hostname}[/dim]")
+        config_table.add_row("Queues", f"[magenta]{', '.join(self.queues)}[/magenta]")
+        config_table.add_row("Concurrency", f"[green]{self.concurrency}[/green]")
+        config_table.add_row("Driver", f"[yellow]{type(self.queue_driver).__name__}[/yellow]")
+
+        if self.max_tasks:
+            config_table.add_row("Max Tasks", f"[orange1]{self.max_tasks}[/orange1]")
+
+        if self.process_pool_size:
+            config_table.add_row("Process Pool", f"[cyan]{self.process_pool_size} workers[/cyan]")
+
+        # Create a nice panel
+        panel = Panel(
+            config_table,
+            title="[bold green]⚡ AsyncTasq Worker Starting[/bold green]",
+            border_style="green",
+            padding=(1, 2),
+        )
+
+        console.print()
+        console.print(panel)
+        console.print()
+
     async def start(self) -> None:
         """Initialize worker and begin processing tasks until shutdown.
 
@@ -95,6 +133,9 @@ async def start(self) -> None:
         except ImportError:
             logger.info("uvloop not available, using default event loop policy")
 
+        # Display startup banner
+        self._display_startup_banner()
+
         self._running = True
         self._start_time = datetime.now(UTC)
 
@@ -127,13 +168,6 @@ async def start(self) -> None:
         for sig in (signal.SIGTERM, signal.SIGINT):
             loop.add_signal_handler(sig, self._handle_shutdown)
 
-        logger.info(
-            "Worker %s starting: queues=%s, concurrency=%d",
-            self.worker_id,
-            self.queues,
-            self.concurrency,
-        )
-
         # Emit worker_online event via global emitters
         await EventRegistry.emit(
             WorkerEvent(
diff --git a/src/asynctasq/monitoring/emitters.py b/src/asynctasq/monitoring/emitters.py
@@ -45,6 +45,40 @@ class LoggingEventEmitter(EventEmitter):
     Uses Rich for colorized, styled console output with icons and visual hierarchy.
     """
 
+    def _format_duration(self, duration_ms: int | None) -> str:
+        """Format duration with color-coded performance indicators."""
+        if duration_ms is None:
+            return ""
+
+        # Convert to seconds for readability
+        duration_s = duration_ms / 1000.0
+
+        # Color-code based on performance
+        if duration_s < 1.0:
+            color = "green"
+            icon = "⚡"
+        elif duration_s < 5.0:
+            color = "cyan"
+            icon = "✨"
+        elif duration_s < 30.0:
+            color = "yellow"
+            icon = "⏱️"
+        else:
+            color = "red"
+            icon = "🐌"
+
+        # Format with appropriate precision
+        if duration_s < 1.0:
+            duration_str = f"{duration_ms}ms"
+        elif duration_s < 60.0:
+            duration_str = f"{duration_s:.2f}s"
+        else:
+            minutes = int(duration_s // 60)
+            seconds = duration_s % 60
+            duration_str = f"{minutes}m {seconds:.1f}s"
+
+        return f"{icon} [{color}]{duration_str}[/{color}]"
+
     def _format_task_event(self, event: TaskEvent) -> str:
         """Format a task event with colors and icons."""
         # Event type to emoji/icon mapping
@@ -53,18 +87,37 @@ def _format_task_event(self, event: TaskEvent) -> str:
             "task_completed": "✅",
             "task_failed": "❌",
             "task_retrying": "🔄",
+            "task_reenqueued": "📤",
+            "task_enqueued": "📥",
+            "task_cancelled": "🚫",
         }
 
         icon = event_icons.get(event.event_type.value, "📋")
         event_name = event.event_type.value.replace("_", " ").title()
 
-        return (
+        # Base message with event type and task info
+        base_msg = (
             f"{icon} [bold cyan]{event_name}[/bold cyan] "
             f"[dim]task=[/dim][yellow]{event.task_id[:8]}[/yellow] "
-            f"[dim]queue=[/dim][magenta]{event.queue}[/magenta] "
-            f"[dim]worker=[/dim][blue]{event.worker_id}[/blue]"
+            f"[dim]name=[/dim][bold magenta]{event.task_name}[/bold magenta] "
+            f"[dim]queue=[/dim][magenta]{event.queue}[/magenta]"
         )
 
+        # Add attempt info if > 1
+        if event.attempt > 1:
+            base_msg += f" [dim]attempt=[/dim][orange1]{event.attempt}[/orange1]"
+
+        # Add duration for completed/failed tasks
+        if event.duration_ms is not None:
+            duration_str = self._format_duration(event.duration_ms)
+            base_msg += f" [dim]duration=[/dim]{duration_str}"
+
+        # Add error info for failed tasks
+        if event.error and event.event_type.value == "task_failed":
+            base_msg += f"\n  [dim]└─[/dim] [red]Error:[/red] {event.error}"
+
+        return base_msg
+
     def _format_worker_event(self, event: WorkerEvent) -> str:
         """Format a worker event with colors and icons."""
         # Event type to emoji/icon mapping
@@ -77,8 +130,36 @@ def _format_worker_event(self, event: WorkerEvent) -> str:
         icon = event_icons.get(event.event_type.value, "⚙️")
         event_name = event.event_type.value.replace("_", " ").title()
 
+        # Special formatting for worker_online - make it stand out
+        if event.event_type.value == "worker_online":
+            # Create a simple but elegant startup message
+            queues_str = ", ".join(event.queues)
+            return (
+                f"{icon} [bold green]{event_name}[/bold green] "
+                f"[dim]worker=[/dim][bold blue]{event.worker_id}[/bold blue] "
+                f"[dim]queues=[/dim][cyan]\\[{queues_str}][/cyan] "
+                f"[dim]hostname=[/dim][dim]{event.hostname}[/dim]"
+            )
+
+        # Special formatting for worker_offline
+        if event.event_type.value == "worker_offline":
+            uptime_str = ""
+            if event.uptime_seconds:
+                hours = event.uptime_seconds // 3600
+                minutes = (event.uptime_seconds % 3600) // 60
+                seconds = event.uptime_seconds % 60
+                uptime_str = f" [dim]uptime=[/dim][cyan]{hours}h {minutes}m {seconds}s[/cyan]"
+
+            return (
+                f"{icon} [bold red]{event_name}[/bold red] "
+                f"[dim]worker=[/dim][blue]{event.worker_id}[/blue] "
+                f"[dim]processed=[/dim][green]{event.processed}[/green]"
+                f"{uptime_str}"
+            )
+
+        # Standard heartbeat formatting
         return (
-            f"{icon} [bold green]{event_name}[/bold green] "
+            f"{icon} [dim]{event_name}[/dim] "
             f"[dim]worker=[/dim][blue]{event.worker_id}[/blue] "
             f"[dim]active=[/dim][cyan]{event.active}[/cyan] "
             f"[dim]processed=[/dim][green]{event.processed}[/green]"
diff --git a/tests/unit/core/test_worker.py b/tests/unit/core/test_worker.py
@@ -250,12 +250,11 @@ async def test_start_logs_info_message(self) -> None:
             except asyncio.CancelledError:
                 pass
 
-        # Assert
+        # Assert - Worker logs uvloop availability messages
         mock_logger.info.assert_called()
-        # Check that the log message includes queue and concurrency info
         log_calls = [str(call) for call in mock_logger.info.call_args_list]
-        assert any("queues=" in str(call) for call in log_calls)
-        assert any("concurrency=" in str(call) for call in log_calls)
+        # Check that we log something about uvloop or event loop
+        assert any("event loop" in str(call).lower() for call in log_calls)
 
 
 @mark.unit
diff --git a/tests/unit/monitoring/test_emitters.py b/tests/unit/monitoring/test_emitters.py
@@ -60,8 +60,8 @@ async def test_emit_task_event_logs(self, sample_task_event: TaskEvent, caplog)
         assert "🚀" in caplog.text  # Task started icon
         assert "Task Started" in caplog.text  # Formatted event name
         assert "test-tas" in caplog.text  # Truncated task ID (first 8 chars)
+        assert "TestTask" in caplog.text  # Task name
         assert "default" in caplog.text  # Queue name
-        assert "worker-abc123" in caplog.text  # Worker ID
 
     @mark.asyncio
     async def test_emit_worker_event_logs(self, sample_worker_event: WorkerEvent, caplog) -> None:
@@ -76,8 +76,8 @@ async def test_emit_worker_event_logs(self, sample_worker_event: WorkerEvent, ca
         assert "🟢" in caplog.text  # Worker online icon
         assert "Worker Online" in caplog.text  # Formatted event name
         assert "worker-abc123" in caplog.text  # Worker ID
-        assert "5" in caplog.text  # Active tasks
-        assert "100" in caplog.text  # Processed count
+        assert "default" in caplog.text  # Queue names
+        assert "test-host" in caplog.text  # Hostname
 
     @mark.asyncio
     async def test_close_is_noop(self) -> None:
