| | |
- vcorelib.task.Task(vcorelib.logging.LoggerMixin)
-
- SubprocessLogMixin
-
- SubprocessExec
- SubprocessExecStreamed
- SubprocessShell
- SubprocessShellStreamed
class SubprocessExec(SubprocessLogMixin) |
| |
SubprocessExec(name: 'str', *args, execute: 'TaskExecute' = None, log: 'Logger' = None, timer: 'Timer' = None, target: 'Target' = None, **kwargs) -> 'None'
A task wrapping a subprocess.
|
| |
- Method resolution order:
- SubprocessExec
- SubprocessLogMixin
- vcorelib.task.Task
- vcorelib.logging.LoggerMixin
- builtins.object
Methods defined here:
- async run(self, inbox: Dict[str, Dict[str, Any]], outbox: Dict[str, Any], *caller_args, args: str = '--version', program: str = '/home/vkottler/src/vkottler/workspace/vcorelib/venv3.12/bin/python', separator: str = '::', require_success: bool = True, **kwargs) -> bool
- Create a subprocess, wait for it to exit and add results to the outbox.
Data and other attributes defined here:
- __annotations__ = {}
Methods inherited from SubprocessLogMixin:
- async exec(self, program: str, *caller_args: str, args: str = '', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> bool
- Execute a command and return whether or not it succeeded.
- async shell(self, cmd: str, *caller_args: str, args: str = '', joiner: str = ' ', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> bool
- Execute a shell command and return whether or not it succeeded.
- async shell_cmd_in_dir(self, path: pathlib.Path, cmd: List[str], joiner: str = ' ', cd: str = 'cd', **kwargs) -> bool
- Run a shell command in a specific directory.
- async subprocess_exec(self, program: str, *caller_args: str, args: str = '', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> asyncio.subprocess.Process
- Create a process using subprocess exec and log what the arguments were.
- async subprocess_shell(self, cmd: str, *caller_args: str, args: str = '', joiner: str = ' ', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> asyncio.subprocess.Process
- Create a process using subprocess shell and log what the command is.
Methods inherited from vcorelib.task.Task:
- __init__(self, name: 'str', *args, execute: 'TaskExecute' = None, log: 'Logger' = None, timer: 'Timer' = None, target: 'Target' = None, **kwargs) -> 'None'
- Initialize this task.
- __str__(self) -> 'str'
- Convert this task into a string.
- create_execute(self, *args, **kwargs) -> 'TaskExecute'
- Create a default execute function for this task.
- depend_on(self, task: "'Task'", eloop: '_asyncio.AbstractEventLoop' = None, **kwargs) -> 'bool'
- Register other tasks' output data to your input box. Return true
if a new dependency was added.
- depend_on_all(self, tasks: "_Iterable['Task']", **kwargs) -> 'int'
- Register multiple dependencies. Return the number of dependencies
added.
- async dispatch(self, caller: "'Task'" = None, init_only: 'bool' = False, substitutions: 'Substitutions' = None, **kwargs) -> 'None'
- Dispatch this task and return whether or not it succeeded.
- dispatch_init(self, caller: "'Task'" = None, substitutions: 'Substitutions' = None, **kwargs) -> '_Tuple[Substitutions, str]'
- Perform some dispatch initialization.
- resolve(self, log: 'Logger', compiled: 'str', substitutions: 'Substitutions' = None) -> 'None'
- Mark this task resolved.
- async resolve_dependencies(self, **substitutions) -> 'None'
- A default dependency resolver for tasks. Note that the default
dependency resolver cannot propagate current-task substitutions to
its dependencies as they've already been explicitly registered.
- resolved(self, compiled: 'str', substitutions: 'Substitutions' = None) -> 'bool'
- Override this in a derived task to implement more complex logic.
- async run_enter(self, _inbox: 'Inbox', _outbox: 'Outbox', *_args, **_kwargs) -> 'bool'
- A default enter method.
- async run_exit(self, _inbox: 'Inbox', _outbox: 'Outbox', *_args, **_kwargs) -> 'bool'
- A default exit method.
- task_fail(self, compiled: 'str', substitutions: 'Substitutions', caller: "'Task'" = None, indent: 'int' = 4) -> 'TaskFailed'
- Build an exception message for when this task fails.
Readonly properties inherited from vcorelib.task.Task:
- stack
- Get this task's execution stack.
Data and other attributes inherited from vcorelib.task.Task:
- default_requirements = set()
- stages = ['run_enter', 'run', 'run_exit']
Methods inherited from vcorelib.logging.LoggerMixin:
- log_time(self, message: str, *args, level: int = 20, reminder: bool = False, **kwargs) -> Iterator[NoneType]
- A simple wrapper.
Data descriptors inherited from vcorelib.logging.LoggerMixin:
- __dict__
- dictionary for instance variables
- __weakref__
- list of weak references to the object
|
class SubprocessExecStreamed(SubprocessLogMixin) |
| |
SubprocessExecStreamed(name: 'str', *args, execute: 'TaskExecute' = None, log: 'Logger' = None, timer: 'Timer' = None, target: 'Target' = None, **kwargs) -> 'None'
A task wrapping a subprocess.
|
| |
- Method resolution order:
- SubprocessExecStreamed
- SubprocessLogMixin
- vcorelib.task.Task
- vcorelib.logging.LoggerMixin
- builtins.object
Methods defined here:
- async run(self, inbox: Dict[str, Dict[str, Any]], outbox: Dict[str, Any], *caller_args, args: str = '--version', program: str = '/home/vkottler/src/vkottler/workspace/vcorelib/venv3.12/bin/python', separator: str = '::', require_success: bool = True, **kwargs) -> bool
- Create a subprocess, wait for it to exit and add results to the outbox.
Data and other attributes defined here:
- __annotations__ = {}
Methods inherited from SubprocessLogMixin:
- async exec(self, program: str, *caller_args: str, args: str = '', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> bool
- Execute a command and return whether or not it succeeded.
- async shell(self, cmd: str, *caller_args: str, args: str = '', joiner: str = ' ', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> bool
- Execute a shell command and return whether or not it succeeded.
- async shell_cmd_in_dir(self, path: pathlib.Path, cmd: List[str], joiner: str = ' ', cd: str = 'cd', **kwargs) -> bool
- Run a shell command in a specific directory.
- async subprocess_exec(self, program: str, *caller_args: str, args: str = '', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> asyncio.subprocess.Process
- Create a process using subprocess exec and log what the arguments were.
- async subprocess_shell(self, cmd: str, *caller_args: str, args: str = '', joiner: str = ' ', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> asyncio.subprocess.Process
- Create a process using subprocess shell and log what the command is.
Methods inherited from vcorelib.task.Task:
- __init__(self, name: 'str', *args, execute: 'TaskExecute' = None, log: 'Logger' = None, timer: 'Timer' = None, target: 'Target' = None, **kwargs) -> 'None'
- Initialize this task.
- __str__(self) -> 'str'
- Convert this task into a string.
- create_execute(self, *args, **kwargs) -> 'TaskExecute'
- Create a default execute function for this task.
- depend_on(self, task: "'Task'", eloop: '_asyncio.AbstractEventLoop' = None, **kwargs) -> 'bool'
- Register other tasks' output data to your input box. Return true
if a new dependency was added.
- depend_on_all(self, tasks: "_Iterable['Task']", **kwargs) -> 'int'
- Register multiple dependencies. Return the number of dependencies
added.
- async dispatch(self, caller: "'Task'" = None, init_only: 'bool' = False, substitutions: 'Substitutions' = None, **kwargs) -> 'None'
- Dispatch this task and return whether or not it succeeded.
- dispatch_init(self, caller: "'Task'" = None, substitutions: 'Substitutions' = None, **kwargs) -> '_Tuple[Substitutions, str]'
- Perform some dispatch initialization.
- resolve(self, log: 'Logger', compiled: 'str', substitutions: 'Substitutions' = None) -> 'None'
- Mark this task resolved.
- async resolve_dependencies(self, **substitutions) -> 'None'
- A default dependency resolver for tasks. Note that the default
dependency resolver cannot propagate current-task substitutions to
its dependencies as they've already been explicitly registered.
- resolved(self, compiled: 'str', substitutions: 'Substitutions' = None) -> 'bool'
- Override this in a derived task to implement more complex logic.
- async run_enter(self, _inbox: 'Inbox', _outbox: 'Outbox', *_args, **_kwargs) -> 'bool'
- A default enter method.
- async run_exit(self, _inbox: 'Inbox', _outbox: 'Outbox', *_args, **_kwargs) -> 'bool'
- A default exit method.
- task_fail(self, compiled: 'str', substitutions: 'Substitutions', caller: "'Task'" = None, indent: 'int' = 4) -> 'TaskFailed'
- Build an exception message for when this task fails.
Readonly properties inherited from vcorelib.task.Task:
- stack
- Get this task's execution stack.
Data and other attributes inherited from vcorelib.task.Task:
- default_requirements = set()
- stages = ['run_enter', 'run', 'run_exit']
Methods inherited from vcorelib.logging.LoggerMixin:
- log_time(self, message: str, *args, level: int = 20, reminder: bool = False, **kwargs) -> Iterator[NoneType]
- A simple wrapper.
Data descriptors inherited from vcorelib.logging.LoggerMixin:
- __dict__
- dictionary for instance variables
- __weakref__
- list of weak references to the object
|
class SubprocessLogMixin(vcorelib.task.Task) |
| |
SubprocessLogMixin(name: 'str', *args, execute: 'TaskExecute' = None, log: 'Logger' = None, timer: 'Timer' = None, target: 'Target' = None, **kwargs) -> 'None'
A class for creating asyncio subprocesses and logging what gets created.
|
| |
- Method resolution order:
- SubprocessLogMixin
- vcorelib.task.Task
- vcorelib.logging.LoggerMixin
- builtins.object
Methods defined here:
- async exec(self, program: str, *caller_args: str, args: str = '', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> bool
- Execute a command and return whether or not it succeeded.
- async shell(self, cmd: str, *caller_args: str, args: str = '', joiner: str = ' ', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> bool
- Execute a shell command and return whether or not it succeeded.
- async shell_cmd_in_dir(self, path: pathlib.Path, cmd: List[str], joiner: str = ' ', cd: str = 'cd', **kwargs) -> bool
- Run a shell command in a specific directory.
- async subprocess_exec(self, program: str, *caller_args: str, args: str = '', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> asyncio.subprocess.Process
- Create a process using subprocess exec and log what the arguments were.
- async subprocess_shell(self, cmd: str, *caller_args: str, args: str = '', joiner: str = ' ', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> asyncio.subprocess.Process
- Create a process using subprocess shell and log what the command is.
Data and other attributes defined here:
- __annotations__ = {}
Methods inherited from vcorelib.task.Task:
- __init__(self, name: 'str', *args, execute: 'TaskExecute' = None, log: 'Logger' = None, timer: 'Timer' = None, target: 'Target' = None, **kwargs) -> 'None'
- Initialize this task.
- __str__(self) -> 'str'
- Convert this task into a string.
- create_execute(self, *args, **kwargs) -> 'TaskExecute'
- Create a default execute function for this task.
- depend_on(self, task: "'Task'", eloop: '_asyncio.AbstractEventLoop' = None, **kwargs) -> 'bool'
- Register other tasks' output data to your input box. Return true
if a new dependency was added.
- depend_on_all(self, tasks: "_Iterable['Task']", **kwargs) -> 'int'
- Register multiple dependencies. Return the number of dependencies
added.
- async dispatch(self, caller: "'Task'" = None, init_only: 'bool' = False, substitutions: 'Substitutions' = None, **kwargs) -> 'None'
- Dispatch this task and return whether or not it succeeded.
- dispatch_init(self, caller: "'Task'" = None, substitutions: 'Substitutions' = None, **kwargs) -> '_Tuple[Substitutions, str]'
- Perform some dispatch initialization.
- resolve(self, log: 'Logger', compiled: 'str', substitutions: 'Substitutions' = None) -> 'None'
- Mark this task resolved.
- async resolve_dependencies(self, **substitutions) -> 'None'
- A default dependency resolver for tasks. Note that the default
dependency resolver cannot propagate current-task substitutions to
its dependencies as they've already been explicitly registered.
- resolved(self, compiled: 'str', substitutions: 'Substitutions' = None) -> 'bool'
- Override this in a derived task to implement more complex logic.
- async run(self, inbox: 'Inbox', outbox: 'Outbox', *args, **kwargs) -> 'bool'
- Override this method to implement the task.
- async run_enter(self, _inbox: 'Inbox', _outbox: 'Outbox', *_args, **_kwargs) -> 'bool'
- A default enter method.
- async run_exit(self, _inbox: 'Inbox', _outbox: 'Outbox', *_args, **_kwargs) -> 'bool'
- A default exit method.
- task_fail(self, compiled: 'str', substitutions: 'Substitutions', caller: "'Task'" = None, indent: 'int' = 4) -> 'TaskFailed'
- Build an exception message for when this task fails.
Readonly properties inherited from vcorelib.task.Task:
- stack
- Get this task's execution stack.
Data and other attributes inherited from vcorelib.task.Task:
- default_requirements = set()
- stages = ['run_enter', 'run', 'run_exit']
Methods inherited from vcorelib.logging.LoggerMixin:
- log_time(self, message: str, *args, level: int = 20, reminder: bool = False, **kwargs) -> Iterator[NoneType]
- A simple wrapper.
Data descriptors inherited from vcorelib.logging.LoggerMixin:
- __dict__
- dictionary for instance variables
- __weakref__
- list of weak references to the object
|
class SubprocessShell(SubprocessLogMixin) |
| |
SubprocessShell(name: 'str', *args, execute: 'TaskExecute' = None, log: 'Logger' = None, timer: 'Timer' = None, target: 'Target' = None, **kwargs) -> 'None'
A task wrapping a shell command.
|
| |
- Method resolution order:
- SubprocessShell
- SubprocessLogMixin
- vcorelib.task.Task
- vcorelib.logging.LoggerMixin
- builtins.object
Methods defined here:
- async run(self, inbox: Dict[str, Dict[str, Any]], outbox: Dict[str, Any], *caller_args, args: str = '', cmd: str = '/home/vkottler/src/vkottler/workspace/vcorelib/venv3.12/bin/python --version', joiner: str = ' ', separator: str = '::', require_success: bool = True, **kwargs) -> bool
- Run a shell command, wait for it to exit and add results to the outbox.
Data and other attributes defined here:
- __annotations__ = {}
Methods inherited from SubprocessLogMixin:
- async exec(self, program: str, *caller_args: str, args: str = '', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> bool
- Execute a command and return whether or not it succeeded.
- async shell(self, cmd: str, *caller_args: str, args: str = '', joiner: str = ' ', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> bool
- Execute a shell command and return whether or not it succeeded.
- async shell_cmd_in_dir(self, path: pathlib.Path, cmd: List[str], joiner: str = ' ', cd: str = 'cd', **kwargs) -> bool
- Run a shell command in a specific directory.
- async subprocess_exec(self, program: str, *caller_args: str, args: str = '', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> asyncio.subprocess.Process
- Create a process using subprocess exec and log what the arguments were.
- async subprocess_shell(self, cmd: str, *caller_args: str, args: str = '', joiner: str = ' ', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> asyncio.subprocess.Process
- Create a process using subprocess shell and log what the command is.
Methods inherited from vcorelib.task.Task:
- __init__(self, name: 'str', *args, execute: 'TaskExecute' = None, log: 'Logger' = None, timer: 'Timer' = None, target: 'Target' = None, **kwargs) -> 'None'
- Initialize this task.
- __str__(self) -> 'str'
- Convert this task into a string.
- create_execute(self, *args, **kwargs) -> 'TaskExecute'
- Create a default execute function for this task.
- depend_on(self, task: "'Task'", eloop: '_asyncio.AbstractEventLoop' = None, **kwargs) -> 'bool'
- Register other tasks' output data to your input box. Return true
if a new dependency was added.
- depend_on_all(self, tasks: "_Iterable['Task']", **kwargs) -> 'int'
- Register multiple dependencies. Return the number of dependencies
added.
- async dispatch(self, caller: "'Task'" = None, init_only: 'bool' = False, substitutions: 'Substitutions' = None, **kwargs) -> 'None'
- Dispatch this task and return whether or not it succeeded.
- dispatch_init(self, caller: "'Task'" = None, substitutions: 'Substitutions' = None, **kwargs) -> '_Tuple[Substitutions, str]'
- Perform some dispatch initialization.
- resolve(self, log: 'Logger', compiled: 'str', substitutions: 'Substitutions' = None) -> 'None'
- Mark this task resolved.
- async resolve_dependencies(self, **substitutions) -> 'None'
- A default dependency resolver for tasks. Note that the default
dependency resolver cannot propagate current-task substitutions to
its dependencies as they've already been explicitly registered.
- resolved(self, compiled: 'str', substitutions: 'Substitutions' = None) -> 'bool'
- Override this in a derived task to implement more complex logic.
- async run_enter(self, _inbox: 'Inbox', _outbox: 'Outbox', *_args, **_kwargs) -> 'bool'
- A default enter method.
- async run_exit(self, _inbox: 'Inbox', _outbox: 'Outbox', *_args, **_kwargs) -> 'bool'
- A default exit method.
- task_fail(self, compiled: 'str', substitutions: 'Substitutions', caller: "'Task'" = None, indent: 'int' = 4) -> 'TaskFailed'
- Build an exception message for when this task fails.
Readonly properties inherited from vcorelib.task.Task:
- stack
- Get this task's execution stack.
Data and other attributes inherited from vcorelib.task.Task:
- default_requirements = set()
- stages = ['run_enter', 'run', 'run_exit']
Methods inherited from vcorelib.logging.LoggerMixin:
- log_time(self, message: str, *args, level: int = 20, reminder: bool = False, **kwargs) -> Iterator[NoneType]
- A simple wrapper.
Data descriptors inherited from vcorelib.logging.LoggerMixin:
- __dict__
- dictionary for instance variables
- __weakref__
- list of weak references to the object
|
class SubprocessShellStreamed(SubprocessLogMixin) |
| |
SubprocessShellStreamed(name: 'str', *args, execute: 'TaskExecute' = None, log: 'Logger' = None, timer: 'Timer' = None, target: 'Target' = None, **kwargs) -> 'None'
A task wrapping a shell command.
|
| |
- Method resolution order:
- SubprocessShellStreamed
- SubprocessLogMixin
- vcorelib.task.Task
- vcorelib.logging.LoggerMixin
- builtins.object
Methods defined here:
- async run(self, inbox: Dict[str, Dict[str, Any]], outbox: Dict[str, Any], *caller_args, args: str = '', cmd: str = '/home/vkottler/src/vkottler/workspace/vcorelib/venv3.12/bin/python --version', joiner: str = ' ', separator: str = '::', require_success: bool = True, **kwargs) -> bool
- Run a shell command, wait for it to exit and add results to the outbox.
Data and other attributes defined here:
- __annotations__ = {}
Methods inherited from SubprocessLogMixin:
- async exec(self, program: str, *caller_args: str, args: str = '', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> bool
- Execute a command and return whether or not it succeeded.
- async shell(self, cmd: str, *caller_args: str, args: str = '', joiner: str = ' ', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> bool
- Execute a shell command and return whether or not it succeeded.
- async shell_cmd_in_dir(self, path: pathlib.Path, cmd: List[str], joiner: str = ' ', cd: str = 'cd', **kwargs) -> bool
- Run a shell command in a specific directory.
- async subprocess_exec(self, program: str, *caller_args: str, args: str = '', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> asyncio.subprocess.Process
- Create a process using subprocess exec and log what the arguments were.
- async subprocess_shell(self, cmd: str, *caller_args: str, args: str = '', joiner: str = ' ', separator: str = '::', stdout: int = None, stderr: int = None, **kwargs) -> asyncio.subprocess.Process
- Create a process using subprocess shell and log what the command is.
Methods inherited from vcorelib.task.Task:
- __init__(self, name: 'str', *args, execute: 'TaskExecute' = None, log: 'Logger' = None, timer: 'Timer' = None, target: 'Target' = None, **kwargs) -> 'None'
- Initialize this task.
- __str__(self) -> 'str'
- Convert this task into a string.
- create_execute(self, *args, **kwargs) -> 'TaskExecute'
- Create a default execute function for this task.
- depend_on(self, task: "'Task'", eloop: '_asyncio.AbstractEventLoop' = None, **kwargs) -> 'bool'
- Register other tasks' output data to your input box. Return true
if a new dependency was added.
- depend_on_all(self, tasks: "_Iterable['Task']", **kwargs) -> 'int'
- Register multiple dependencies. Return the number of dependencies
added.
- async dispatch(self, caller: "'Task'" = None, init_only: 'bool' = False, substitutions: 'Substitutions' = None, **kwargs) -> 'None'
- Dispatch this task and return whether or not it succeeded.
- dispatch_init(self, caller: "'Task'" = None, substitutions: 'Substitutions' = None, **kwargs) -> '_Tuple[Substitutions, str]'
- Perform some dispatch initialization.
- resolve(self, log: 'Logger', compiled: 'str', substitutions: 'Substitutions' = None) -> 'None'
- Mark this task resolved.
- async resolve_dependencies(self, **substitutions) -> 'None'
- A default dependency resolver for tasks. Note that the default
dependency resolver cannot propagate current-task substitutions to
its dependencies as they've already been explicitly registered.
- resolved(self, compiled: 'str', substitutions: 'Substitutions' = None) -> 'bool'
- Override this in a derived task to implement more complex logic.
- async run_enter(self, _inbox: 'Inbox', _outbox: 'Outbox', *_args, **_kwargs) -> 'bool'
- A default enter method.
- async run_exit(self, _inbox: 'Inbox', _outbox: 'Outbox', *_args, **_kwargs) -> 'bool'
- A default exit method.
- task_fail(self, compiled: 'str', substitutions: 'Substitutions', caller: "'Task'" = None, indent: 'int' = 4) -> 'TaskFailed'
- Build an exception message for when this task fails.
Readonly properties inherited from vcorelib.task.Task:
- stack
- Get this task's execution stack.
Data and other attributes inherited from vcorelib.task.Task:
- default_requirements = set()
- stages = ['run_enter', 'run', 'run_exit']
Methods inherited from vcorelib.logging.LoggerMixin:
- log_time(self, message: str, *args, level: int = 20, reminder: bool = False, **kwargs) -> Iterator[NoneType]
- A simple wrapper.
Data descriptors inherited from vcorelib.logging.LoggerMixin:
- __dict__
- dictionary for instance variables
- __weakref__
- list of weak references to the object
| |