Skip to content

JobManager

JobManager

Bases: Widget

Manage currently running jobs.

Source code in src/textual_shell/widgets/job_manager.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class JobManager(Widget):
    """Manage currently running jobs."""

    DEFAULT_CSS = """
        JobManager {
            border: solid white;

            DataTable {
                border-top: solid white;
            }
        }

    """

    job_list: dict[str, Job] = {}

    def compose(self) -> ComposeResult:
        yield Label('Job Manager')
        yield DataTable()

    def on_mount(self) -> None:
        table = self.query_one(DataTable)
        self.column_keys = table.add_columns('Jobs', 'Status')

    def add_job(
        self,
        job: Annotated[Job, 'The job to add.']
    ) -> None:
        """
        Add a new job to the table and dictionary.

        Args:
            job (Job): The job to add.
        """
        self.job_list[job.id] = job
        table = self.query_one(DataTable)
        row = (job.id, job.status)
        table.add_row(*row, key=job.id)

    def remove_job(
        self,
        job_id: Annotated[str, 'The id of the job.']
    ) -> None:
        """
        Remove a job from the table and dictionary.

        Args:
            job_id (str): The id of the job.
        """
        job = self.job_list.pop(job_id)
        table = self.query_one(DataTable)
        table.remove_row(job_id)

    def update_job_status(
        self,
        job_id: Annotated[str, 'The id of the job'],
        status: Job.Status
    ) -> None:
        """
        Update the status of the job in the table.

        Args:
            job_id (str): The id of the job.
            status (Job.Status): The jobs current status.
        """
        table = self.query_one(DataTable)
        table.update_cell(
            row_key=job_id,
            column_key=self.column_keys[1],
            value=status
        )

    def switch_job_screen(
        self,
        job_id: Annotated[str, 'The id of the job']
    ) -> None:
        """
        Switch to the specified jobs screen if it exists.

        Args:
            job_id (str): The id of the job.
        """
        try:
            job = self.job_list[job_id]
            if job.screen:
                self.app.push_screen(job.screen)

            else:
                self.notify(
                    message=f'{job_id.upper()} has no screen to attach to.',
                    title='Screen Not Found',
                    severity='warning'
                )

        except KeyError as e:
            self.notify(
                message=f'Job with ID: [{job_id.upper()}] does not exist',
                title='Job Not Found',
                severity='error'
            )

    def kill_job(
        self,
        job_id: Annotated[str, 'The id of the job'],
    ) -> None:
        """
        Cancel the job.

        Args:
            job_id (str): The id of the job.
        """
        try:
            job = self.job_list[job_id]
            job.task.cancel()

        except KeyError as e:
            pass

    def on_data_table_cell_selected(
        self,
        event: DataTable.CellSelected
    ) -> None:
        """Attach to the screen of the job."""
        self.switch_job_screen(event.value)

add_job(job)

Add a new job to the table and dictionary.

Parameters:

Name Type Description Default
job Job

The job to add.

 required
Source code in src/textual_shell/widgets/job_manager.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def add_job(
    self,
    job: Annotated[Job, 'The job to add.']
) -> None:
    """
    Add a new job to the table and dictionary.

    Args:
        job (Job): The job to add.
    """
    self.job_list[job.id] = job
    table = self.query_one(DataTable)
    row = (job.id, job.status)
    table.add_row(*row, key=job.id)

kill_job(job_id)

Cancel the job.

Parameters:

Name Type Description Default
job_id str

The id of the job.

 required
Source code in src/textual_shell/widgets/job_manager.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def kill_job(
    self,
    job_id: Annotated[str, 'The id of the job'],
) -> None:
    """
    Cancel the job.

    Args:
        job_id (str): The id of the job.
    """
    try:
        job = self.job_list[job_id]
        job.task.cancel()

    except KeyError as e:
        pass

on_data_table_cell_selected(event)

Attach to the screen of the job.

Source code in src/textual_shell/widgets/job_manager.py 
131
132
133
134
135
136
def on_data_table_cell_selected(
    self,
    event: DataTable.CellSelected
) -> None:
    """Attach to the screen of the job."""
    self.switch_job_screen(event.value)

remove_job(job_id)

Remove a job from the table and dictionary.

Parameters:

Name Type Description Default
job_id str

The id of the job.

 required
Source code in src/textual_shell/widgets/job_manager.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
def remove_job(
    self,
    job_id: Annotated[str, 'The id of the job.']
) -> None:
    """
    Remove a job from the table and dictionary.

    Args:
        job_id (str): The id of the job.
    """
    job = self.job_list.pop(job_id)
    table = self.query_one(DataTable)
    table.remove_row(job_id)

switch_job_screen(job_id)

Switch to the specified jobs screen if it exists.

Parameters:

Name Type Description Default
job_id str

The id of the job.

 required
Source code in src/textual_shell/widgets/job_manager.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def switch_job_screen(
    self,
    job_id: Annotated[str, 'The id of the job']
) -> None:
    """
    Switch to the specified jobs screen if it exists.

    Args:
        job_id (str): The id of the job.
    """
    try:
        job = self.job_list[job_id]
        if job.screen:
            self.app.push_screen(job.screen)

        else:
            self.notify(
                message=f'{job_id.upper()} has no screen to attach to.',
                title='Screen Not Found',
                severity='warning'
            )

    except KeyError as e:
        self.notify(
            message=f'Job with ID: [{job_id.upper()}] does not exist',
            title='Job Not Found',
            severity='error'
        )

update_job_status(job_id, status)

Update the status of the job in the table.

Parameters:

Name Type Description Default
job_id str

The id of the job.

 required
status Status

The jobs current status.

 required
Source code in src/textual_shell/widgets/job_manager.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def update_job_status(
    self,
    job_id: Annotated[str, 'The id of the job'],
    status: Job.Status
) -> None:
    """
    Update the status of the job in the table.

    Args:
        job_id (str): The id of the job.
        status (Job.Status): The jobs current status.
    """
    table = self.query_one(DataTable)
    table.update_cell(
        row_key=job_id,
        column_key=self.column_keys[1],
        value=status
    )