Switch from deprecated getTaskList to listOperations (#14)

.github/workflows/ci.yaml

@@ -8,7 +8,7 @@ jobs:
   test:
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.9", "3.10", "3.11", "3.12"]
 
     runs-on: ubuntu-latest
 

pyproject.toml

@@ -8,7 +8,7 @@ dynamic = ["version"]
 description = "Notifications for Earth Engine tasks."
 readme = "README.md"
 license = { file = "LICENSE" }
-requires-python = ">=3.8"
+requires-python = ">=3.9"
 authors = [{ name = "Aaron Zuspan" }]
 keywords = ["earth-engine", "notifications", "tasks", "cli", "command-line"]
 dependencies = [
@@ -19,6 +19,7 @@ dependencies = [
     "requests",
     "click",
     "rich-click>=1.2.1",
+    "pydantic",
 ]
 
 [project.urls]
@@ -62,4 +63,7 @@ known-first-party = ["taskee"]
 [tool.pytest.ini_options]
 markers = [
     "no_config: mark test to run without a config file",
-]
+]
+
+[tool.ruff.lint.pyupgrade]
+keep-runtime-typing = true

src/taskee/__init__.py

@@ -1 +1,5 @@
+from .taskee import Taskee
+
 __version__ = "0.0.4"
+
+__all__ = ["Taskee"]

src/taskee/cli/cli.py

@@ -5,8 +5,8 @@
 from rich.status import Status
 
 from taskee.cli.commands import dashboard, log, tasks, test
-from taskee.events import EVENT_TYPES, Error
-from taskee.notifiers import NOTIFIER_TYPES
+from taskee.events import ErrorEvent, EventEnum
+from taskee.notifiers import NotifierEnum
 from taskee.taskee import Taskee
 
 click.rich_click.SHOW_ARGUMENTS = True
@@ -20,7 +20,7 @@
     "--private-key",
     default=None,
     type=click.Path(exists=True, dir_okay=False),
-    help="Path to private key file for Earth Engine authentication.",
+    help="Optional path to private key file for Earth Engine authentication.",
 )
 
 NOTIFIERS_OPTION = click.option(
@@ -29,22 +29,26 @@
     "--notifier",
     default=("native",),
     multiple=True,
-    type=click.Choice(list(NOTIFIER_TYPES.keys()) + ["all"], case_sensitive=False),
-    help="One or more notifiers to run (or all).",
+    type=click.Choice(
+        list(NotifierEnum.__members__.keys()) + ["all"], case_sensitive=False
+    ),
+    help="The notifier to run (or all).",
 )
 
 INTERVAL_OPTION = click.option(
     "interval_mins",
     "-i",
-    "--interval_mins",
+    "--interval-mins",
     default=5.0,
     help="Minutes between queries to Earth Engine for task updates.",
 )
 
 WATCH_FOR_ARG = click.argument(
     "watch_for",
     nargs=-1,
-    type=click.Choice(choices=list(EVENT_TYPES.keys()) + ["all"], case_sensitive=False),
+    type=click.Choice(
+        choices=list(EventEnum.__members__.keys()) + ["all"], case_sensitive=False
+    ),
 )
 
 
@@ -94,9 +98,9 @@ def start_command(
     ```
     """
     if "all" in notifiers:
-        notifiers = tuple(NOTIFIER_TYPES.keys())
+        notifiers = tuple(NotifierEnum.__members__.keys())
     if "all" in watch_for:
-        watch_for = tuple(EVENT_TYPES.keys())
+        watch_for = tuple(EventEnum.__members__.keys())
     elif len(watch_for) == 0:
         watch_for = ("completed", "failed", "error")
 
@@ -106,14 +110,14 @@ def start_command(
         credentials = "persistent"
 
     mode_func = modes[mode]
-    t = Taskee(notifiers=notifiers, credentials=credentials)
+    t = Taskee(notifiers=notifiers, watch_for=watch_for, credentials=credentials)
 
     try:
-        mode_func(t, watch_for=watch_for, interval_minutes=interval_mins)
+        mode_func(t, interval_minutes=interval_mins)
     except Exception as e:
         if "error" in [event.lower() for event in watch_for]:
-            event = Error()
-            t.dispatcher.notify(event.title, event.message)
+            t.event_queue.append(ErrorEvent())
+            t.dispatch()
         raise e
     except KeyboardInterrupt:
         return
@@ -131,7 +135,7 @@ def tasks_command(max_tasks: int, private_key: str | None) -> None:
 
     with Status("Retrieving tasks from Earth Engine...", spinner="bouncingBar"):
         t = Taskee(notifiers=tuple(), credentials=credentials)
-        tasks.tasks(t, max_tasks=max_tasks)
+        tasks.tasks(t.tasks, max_tasks=max_tasks)
 
 
 @taskee.command(name="test", short_help="Send test notifications.")
@@ -148,9 +152,10 @@ def test_command(notifiers: tuple[str, ...]) -> None:
     ```
     """
     if "all" in notifiers:
-        notifiers = tuple(NOTIFIER_TYPES.keys())
+        notifiers = tuple(NotifierEnum.__members__.keys())
 
-    test.test(notifiers)
+    notifier_instances = tuple(NotifierEnum[notifier].value() for notifier in notifiers)
+    test.test(notifier_instances)
 
 
 if __name__ == "__main__":

src/taskee/cli/commands/dashboard.py

@@ -13,34 +13,31 @@
 from rich.table import Table
 from rich.text import Text
 
-from taskee import events, states
-from taskee.cli.styles import get_style
+from taskee.cli.commands.tasks import create_task_table
+from taskee.cli.styles import STYLES
 from taskee.taskee import Taskee
 
-REFRESH_SECONDS = 1
-MAX_ROWS = 20
-TABLE_HEADER_HEIGHT = 6
-BOX_STYLE = box.SIMPLE_HEAD
-
 if TYPE_CHECKING:
-    from taskee.events import Event
+    from taskee.events import _Event
+
 
+class _Dashboard:
+    REFRESH_SECONDS = 1
+    MAX_ROWS = 20
+    TABLE_HEADER_HEIGHT = 6
 
-class Dashboard:
     def __init__(
         self,
         t: Taskee,
-        watch_for: tuple[str, ...] = ("error", "completed", "failed"),
         interval_minutes: float = 5.0,
     ):
-        self.event_log: deque[Event] = deque(maxlen=MAX_ROWS)
-        self.last_checked = time.time()
+        self.event_log: deque[_Event] = deque(maxlen=self.MAX_ROWS)
+        self.last_checked = 0.0
 
         self.t = t
-        self.watch_events = set([events.get_event(name) for name in watch_for])
         self.interval_seconds = interval_minutes * 60.0
 
-        self.layout = self.create_layout()
+        self.layout = self._create_layout()
         self.window = Panel(
             self.layout,
             title="[bold white]taskee",
@@ -49,50 +46,50 @@ def __init__(
         )
 
         # Initialize dashboard before starting Live so we don't render a preview layout
-        self.update_display()
+        self._update_display()
 
     @property
-    def elapsed(self) -> float:
+    def _elapsed(self) -> float:
         """Return the time elapsed since the last event update"""
         return time.time() - self.last_checked
 
     @property
-    def time_remaining(self) -> float:
+    def _time_remaining(self) -> float:
         """Return the time remaining until the next event update"""
-        return self.interval_seconds - self.elapsed
+        return self.interval_seconds - self._elapsed
 
-    def run(self) -> None:
+    def _run(self) -> None:
         """Run the dashboard indefinitely."""
         with Live(self.window):
             while True:
-                if self.elapsed > self.interval_seconds:
-                    self.update_events()
+                if self._elapsed > self.interval_seconds:
+                    self._update_events()
 
-                self.update_display()
-                time.sleep(REFRESH_SECONDS)
+                self._update_display()
+                time.sleep(self.REFRESH_SECONDS)
 
-    def update_events(self) -> None:
+    def _update_events(self) -> None:
         """Update Earth Engine tasks and store new events."""
-        self.t._update(self.watch_events)
+        new_events = self.t.update()
+        self.t.dispatch()
 
-        new_events = self.t.manager.events
         for event in new_events:
             self.event_log.appendleft(event)
 
         self.last_checked = time.time()
 
-    def update_display(self) -> None:
+    def _update_display(self) -> None:
         """Update the dasboard display."""
-        self.layout["header"].update(self.create_header())
-        self.layout["progress"].update(self.create_progress())
+        self.layout["header"].update(self._create_header())
+        self.layout["progress"].update(self._create_progress())
 
-        task_table, event_table = self.create_tables()
+        task_table, event_table = self._create_tables()
         self.layout["tasks"].update(task_table)
         self.layout["events"].update(event_table)
-        self.layout["tasks"].size = task_table.row_count + TABLE_HEADER_HEIGHT
-        self.layout["events"].size = event_table.row_count + TABLE_HEADER_HEIGHT
+        self.layout["tasks"].size = task_table.row_count + self.TABLE_HEADER_HEIGHT
+        self.layout["events"].size = event_table.row_count + self.TABLE_HEADER_HEIGHT
 
-    def create_layout(self) -> Layout:
+    def _create_layout(self) -> Layout:
         """Create the dashboard Layout"""
         layout = Layout()
 
@@ -103,74 +100,45 @@ def create_layout(self) -> Layout:
             Layout(name="tasks", minimum_size=4),
         )
 
-        layout["progress"].update(self.create_progress())
-        layout["header"].update(self.create_header())
+        layout["progress"].update(self._create_progress())
+        layout["header"].update(self._create_header())
 
         return layout
 
-    def create_header(self) -> Table:
+    def _create_header(self) -> Table:
         """Create the dashboard Header showing the update time and controls."""
         grid = Table.grid(expand=True)
         grid.add_column(justify="left")
         grid.add_column(justify="right")
         grid.add_row(
             "[italic]Next update in"
-            f" {humanize.naturaldelta(self.time_remaining)}...[/]",
+            f" {humanize.naturaldelta(self._time_remaining)}...[/]",
             Text("Press CTRL + C to exit...", style="dim"),
         )
 
         return grid
 
-    def create_progress(self) -> ProgressBar:
+    def _create_progress(self) -> ProgressBar:
         return ProgressBar(
             total=self.interval_seconds,
-            completed=self.interval_seconds - self.time_remaining,
+            completed=self.interval_seconds - self._time_remaining,
             complete_style="bright_yellow",
         )
 
-    def create_tables(self) -> tuple[Table, Table]:
-        n_tasks = MAX_ROWS - min(max(len(self.event_log), 1), MAX_ROWS // 2)
-        n_events = MAX_ROWS - n_tasks
+    def _create_tables(self) -> tuple[Table, Table]:
+        n_tasks = self.MAX_ROWS - min(max(len(self.event_log), 1), self.MAX_ROWS // 2)
+        n_events = self.MAX_ROWS - n_tasks
 
-        task_table = self.create_task_table(n_tasks)
-        event_table = self.create_event_table(n_events)
+        task_table = create_task_table(self.t.tasks, n_tasks)
+        event_table = self._create_event_table(n_events)
 
         return task_table, event_table
 
-    def create_task_table(self, max_tasks: int | None = None) -> Table:
-        """Create a table of tasks."""
-        t = Table(
-            title="[bold bright_green]Tasks",
-            box=BOX_STYLE,
-            header_style="bright_green",
-            expand=True,
-        )
-
-        t.add_column("State", justify="right")
-        t.add_column("Description", justify="left")
-        t.add_column("Created", justify="right")
-        t.add_column("Elapsed", justify="right")
-
-        for task in self.t.manager.tasks[:max_tasks]:
-            state_style = get_style(task.state)
-            dim_style = "[dim]" if task.state not in states.ACTIVE else ""
-            time_created = humanize.naturaltime(task.time_created)
-            time_elapsed = humanize.naturaldelta(task.time_elapsed)
-
-            t.add_row(
-                f"[{state_style.color}]{task.state}[/] {state_style.emoji}",
-                f"{dim_style}{task.description}",
-                f"{dim_style}{time_created}",
-                f"{dim_style}{time_elapsed}",
-            )
-
-        return t
-
-    def create_event_table(self, max_events: int | None = None) -> Table:
+    def _create_event_table(self, max_events: int | None = None) -> Table:
         """Create a table of events."""
         t = Table(
             title="[bold bright_blue]Events",
-            box=BOX_STYLE,
+            box=box.SIMPLE_HEAD,
             header_style="bright_blue",
             expand=True,
         )
@@ -179,11 +147,13 @@ def create_event_table(self, max_events: int | None = None) -> Table:
         t.add_column("Message", justify="left")
         t.add_column("Time", justify="right")
 
+        now = time.time()
+
         for event in tuple(self.event_log)[:max_events]:
-            event_time = humanize.naturaltime(event.time)
-            event_style = get_style(event.__class__)
-            event_name = event.__class__.__name__
-            muted_style = "[dim]" if event.__class__ not in self.watch_events else ""
+            event_time = humanize.naturaltime(now)
+            event_style = STYLES[event.__class__]
+            event_name = event.__class__.__name__.replace("Event", "")
+            muted_style = "[dim]" if event.__class__ not in self.t.watch_for else ""
             t.add_row(
                 f"[{event_style.color}]{event_name}[/] {event_style.emoji}",
                 f"{muted_style}{event.message}",
@@ -197,11 +167,10 @@ def create_event_table(self, max_events: int | None = None) -> Table:
 
 def start(
     t: Taskee,
-    watch_for: tuple[str, ...] = ("error", "completed", "failed"),
     interval_minutes: float = 5.0,
 ) -> None:
     """Run an indefinite dashboard. This handles scheduling of Earth Engine updates and
     runs a live-updating dashboard of tasks and events as they occur.
     """
-    dashboard = Dashboard(t, watch_for, interval_minutes)
-    dashboard.run()
+    dashboard = _Dashboard(t, interval_minutes)
+    dashboard._run()