Event: CMD_EXECUTE :param ctx: CtxItem :param cmds: commands dict
(self, ctx: CtxItem, cmds: list)
| 81 | data['cmd'].append(self.get_cmd(option)) # append command |
| 82 | |
| 83 | def cmd(self, ctx: CtxItem, cmds: list): |
| 84 | """ |
| 85 | Event: CMD_EXECUTE |
| 86 | |
| 87 | :param ctx: CtxItem |
| 88 | :param cmds: commands dict |
| 89 | """ |
| 90 | from .worker import Worker |
| 91 | |
| 92 | is_cmd = False |
| 93 | my_commands = [] |
| 94 | for item in cmds: |
| 95 | if item["cmd"] in self.allowed_cmds: |
| 96 | my_commands.append(item) |
| 97 | is_cmd = True |
| 98 | |
| 99 | if not is_cmd: |
| 100 | return |
| 101 | |
| 102 | # set state: busy |
| 103 | self.cmd_prepare(ctx, my_commands) |
| 104 | |
| 105 | try: |
| 106 | worker = Worker() |
| 107 | worker.from_defaults(self) |
| 108 | worker.cmds = my_commands |
| 109 | worker.ctx = ctx |
| 110 | |
| 111 | if not self.is_async(ctx): |
| 112 | worker.run() |
| 113 | return |
| 114 | worker.run_async() |
| 115 | |
| 116 | except Exception as e: |
| 117 | self.error(e) |
no test coverage detected