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 = [{"key": task} for task in ctx.app.task_manager.all_results()] 

24 

25 click.echo(ctx.render(data, headers={"key": "Result key"})) 

26 

27 

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

29@pass_context 

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

31 """List pending tasks.""" 

32 data = [] 

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

34 task_str = str(task) 

35 data.append( 

36 { 

37 "name": getattr(task, "name", task_str.split(": ")[0]), 

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

39 } 

40 ) 

41 

42 click.echo(ctx.render(data, headers={"name": "Task Name", "task_id": "Task ID / UUID"})) 

43 

44 

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

46@pass_context 

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

48 """List registered tasks.""" 

49 load_task_modules() 

50 data = [{"class": task} for task in ctx.app.task_manager._registry._registry] 

51 

52 click.echo(ctx.render(data, headers={"class": "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")