Coverage for seedboxsync/cli/commands/cmd_task.py: 100%

1# 

2# Copyright (C) 2015-2026 Guillaume Kulakowski <guillaume@kulakowski.fr> 

3# 

4# For the full copyright and license information, please view the LICENSE 

5# file that was distributed with this source code. 

6# 

7"""All commands related to the task queue manager operations.""" 

8 

9import click 

10from seedboxsync.cli import Context, group, pass_context 

11from seedboxsync.core.taskmanager.utils import load_task_modules 

12 

13 

14@group("task", help="Task operations on task queue management.") # type: ignore[untyped-decorator] 

15def cli() -> None: 

16 """Empty function for Click sub commands.""" 

17 

18 

19@cli.command("result", help="List results in the result store. Allows determining the currently running tasks.") # type: ignore[untyped-decorator] 

20@pass_context 

21def tasks_result(ctx: Context) -> None: 

22 """List all_results() tasks.""" 

23 data = [] 

24 for task in ctx.app.task_manager.all_results(): 

25 data.append([task]) 

26 

27 click.echo(ctx.render(data, headers=["Result key"])) 

28 

29 

30@cli.command("pending", help="List pending tasks.") # type: ignore[untyped-decorator] 

31@pass_context 

32def tasks_pending(ctx: Context) -> None: 

33 """List pending tasks.""" 

34 data = [] 

35 for task in ctx.app.task_manager.pending(): 

36 name = getattr(task, "name", str(task).split(": ")[0]) 

37 task_id = getattr(task, "id", str(task).split(": ")[-1] if ": " in str(task) else str(task)) 

38 data.append([name, task_id]) 

39 

40 click.echo(ctx.render(data, headers=["Task Name", "Task ID / UUID"])) 

41 

42 

43@cli.command("list", help="List registered tasks.") # type: ignore[untyped-decorator] 

44@pass_context 

45def tasks_list(ctx: Context) -> None: 

46 """List registered tasks.""" 

47 load_task_modules() 

48 data = [] 

49 for task in ctx.app.task_manager._registry._registry: 

50 data.append([task]) 

51 

52 click.echo(ctx.render(data, headers=["Class"])) 

53 

54 

55@cli.command("flush", help="Remove all data from the queue, schedule, and result store.") # type: ignore[untyped-decorator] 

56@pass_context 

57def tasks_flush(ctx: Context) -> None: 

58 """Remove all data from the queue.""" 

59 ctx.app.task_manager.flush() 

60 click.echo("Queue flushed") 

61 

62 

63@cli.command("flush-lock", help="Flush any locks that may be held.") # type: ignore[untyped-decorator] 

64@pass_context 

65def tasks_flush_lock(ctx: Context) -> None: 

66 """Flush any locks that may be held.""" 

67 ctx.app.task_manager.flush_locks() 

68 click.echo("Lock flushed") 

69 

70 

71@cli.command("sync-blackhole", help="Launch asynchrone task sync blackhole.") # type: ignore[untyped-decorator] 

72@pass_context 

73def tasks_sync_blackhole(ctx: Context) -> None: 

74 """Launch asynchrone task sync blackhole.""" 

75 with ctx.app.app_context(): 

76 from seedboxsync.core.taskmanager.task.task_sync_blackhole import sync_blackhole 

77 

78 sync_blackhole() 

79 click.echo("Task sync blackhole launched in task manager") 

80 

81 

82@cli.command("sync-seedbox", help="Launch asynchrone task sync seedbox.") # type: ignore[untyped-decorator] 

83@pass_context 

84def tasks_sync_seedbox(ctx: Context) -> None: 

85 """Launch asynchrone task sync seedbox.""" 

86 with ctx.app.app_context(): 

87 from seedboxsync.core.taskmanager.task.task_sync_seedbox import sync_seedbox 

88 

89 sync_seedbox() 

90 click.echo("Task sync seedbox launched in task manager")