Skip to content

JOBS

Attach

Bases: Job

Job to attach to the screen of the selected job.

Source code in src/textual_shell/commands/jobs.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class Attach(Job):
    """Job to attach to the screen of the selected job."""

    class To_Job(Message):
        """Message to attach to the job."""
        def __init__(self, job_id):
            super().__init__()
            self.job_id = job_id

    def __init__(
        self,
        selected_job: Annotated[str, 'The id of the selected job.'],
        *args, **kwargs
    ) -> None:
        super().__init__(*args, **kwargs)
        self.selected_job = selected_job

    async def execute(self):
        """Send request to attach to the selected job."""
        self.running()
        self.shell.post_message(
            self.To_Job(
                self.selected_job
            )
        )
        self.completed()

To_Job

Bases: Message

Message to attach to the job.

Source code in src/textual_shell/commands/jobs.py 
13
14
15
16
17
class To_Job(Message):
    """Message to attach to the job."""
    def __init__(self, job_id):
        super().__init__()
        self.job_id = job_id

execute() async

Send request to attach to the selected job.

Source code in src/textual_shell/commands/jobs.py 
27
28
29
30
31
32
33
34
35
async def execute(self):
    """Send request to attach to the selected job."""
    self.running()
    self.shell.post_message(
        self.To_Job(
            self.selected_job
        )
    )
    self.completed()

Jobs

Bases: Command

Command for interacting with the jobs running in the shell.

Source code in src/textual_shell/commands/jobs.py 
 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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class Jobs(Command):
    """Command for interacting with the jobs running in the shell."""

    DEFINITION = {
        'jobs': CommandNode(
            name='jobs',
            description='Manage jobs.',
            children={
                'attach': CommandNode(
                    name='attach',
                    description="Attach to the job's screen."
                ),
                'kill': CommandNode(
                    name='kill',
                    description='Kill the job.'
                )
            }
        )
    }

    JOBS = []

    def get_suggestions(
        self,
        cmdline: Annotated[list[str], 'The current value of the command line.']
    ) -> Annotated[list[str], 'A list of possible next values']:
        """
        Get a list of suggestions for autocomplete via the current args neighbors.

        Args:
            cmdline (list[str]): The current value of the  command line.

        Returns:
            suggestions (List[str]): List of current node's neighbors names.
        """
        if len(cmdline) == 2:
            if cmdline[1] == 'kill' or cmdline[1] == 'attach':
                return self.JOBS

        else:
            return super().get_suggestions(cmdline)

    def add_job_id(self, job_id: str) -> None:
        """
        Add the job id to use for suggestions.

        Args:
            job_id (str): Tht job to remove.
        """
        self.JOBS.append(job_id)

    def remove_job_id(self, job_id: str) -> None:
        """
        Remove the job id from the suggestions.

        Args:
            job_id (str): Tht job to remove.
        """
        self.JOBS.remove(job_id)

    def create_job(self, *args) -> Attach | Kill:
        """Create the job to manage other jobs."""
        if len(args) != 2:
            self.send_log('Invalid args', logging.ERROR)
            return

        if args[0] == 'attach':
            return Attach(
                selected_job=args[1],
                shell=self.shell,
                cmd=self.name
            )

        elif args[0] == 'kill':
            return Kill(
                selected_job=args[1],
                shell=self.shell,
                cmd=self.name
            )

        else:
            self.shell.notify(
                message='Invalid subcommand.',
                title='Command: jobs',
                severity='error'
            )

add_job_id(job_id)

Add the job id to use for suggestions.

Parameters:

Name Type Description Default
job_id str

Tht job to remove.

 required
Source code in src/textual_shell/commands/jobs.py 
108
109
110
111
112
113
114
115
def add_job_id(self, job_id: str) -> None:
    """
    Add the job id to use for suggestions.

    Args:
        job_id (str): Tht job to remove.
    """
    self.JOBS.append(job_id)

create_job(*args)

Create the job to manage other jobs.

Source code in src/textual_shell/commands/jobs.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def create_job(self, *args) -> Attach | Kill:
    """Create the job to manage other jobs."""
    if len(args) != 2:
        self.send_log('Invalid args', logging.ERROR)
        return

    if args[0] == 'attach':
        return Attach(
            selected_job=args[1],
            shell=self.shell,
            cmd=self.name
        )

    elif args[0] == 'kill':
        return Kill(
            selected_job=args[1],
            shell=self.shell,
            cmd=self.name
        )

    else:
        self.shell.notify(
            message='Invalid subcommand.',
            title='Command: jobs',
            severity='error'
        )

get_suggestions(cmdline)

Get a list of suggestions for autocomplete via the current args neighbors.

Parameters:

Name Type Description Default
cmdline list[str]

The current value of the command line.

 required

Returns:

Name Type Description
suggestions List[str]

List of current node's neighbors names.
Source code in src/textual_shell/commands/jobs.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def get_suggestions(
    self,
    cmdline: Annotated[list[str], 'The current value of the command line.']
) -> Annotated[list[str], 'A list of possible next values']:
    """
    Get a list of suggestions for autocomplete via the current args neighbors.

    Args:
        cmdline (list[str]): The current value of the  command line.

    Returns:
        suggestions (List[str]): List of current node's neighbors names.
    """
    if len(cmdline) == 2:
        if cmdline[1] == 'kill' or cmdline[1] == 'attach':
            return self.JOBS

    else:
        return super().get_suggestions(cmdline)

remove_job_id(job_id)

Remove the job id from the suggestions.

Parameters:

Name Type Description Default
job_id str

Tht job to remove.

 required
Source code in src/textual_shell/commands/jobs.py 
117
118
119
120
121
122
123
124
def remove_job_id(self, job_id: str) -> None:
    """
    Remove the job id from the suggestions.

    Args:
        job_id (str): Tht job to remove.
    """
    self.JOBS.remove(job_id)

Kill

Bases: Job

Job to kill another job.

Source code in src/textual_shell/commands/jobs.py 
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
class Kill(Job):
    """Job to kill another job."""

    class Selected(Message):
        """Message to kill the selected job."""
        def __init__(self, job_id):
            super().__init__()
            self.job_id = job_id

    def __init__(
        self,
        selected_job: Annotated[str, 'The id of the selected job.'],
        *args, **kwargs
    ) -> None:
        super().__init__(*args, **kwargs)
        self.selected_job = selected_job

    async def execute(self):
        """Send the request to kill the selected job."""
        self.running()
        self.shell.post_message(
            self.Selected(
                self.selected_job
            )
        )
        self.completed()

Selected

Bases: Message

Message to kill the selected job.

Source code in src/textual_shell/commands/jobs.py 
41
42
43
44
45
class Selected(Message):
    """Message to kill the selected job."""
    def __init__(self, job_id):
        super().__init__()
        self.job_id = job_id

execute() async

Send the request to kill the selected job.

Source code in src/textual_shell/commands/jobs.py 
55
56
57
58
59
60
61
62
63
async def execute(self):
    """Send the request to kill the selected job."""
    self.running()
    self.shell.post_message(
        self.Selected(
            self.selected_job
        )
    )
    self.completed()