diff --git a/README.md b/README.md index 8e992f3..cd5f587 100644 --- a/README.md +++ b/README.md @@ -91,8 +91,8 @@ WHENDY WORLD $ls .github/* ``` -### 10. Static interpolation -Denoted by {{variable_or_function_call_here}}. For static interpolation, no quotes, spaces or expressions within the {{}} or in the string being injected. +### 10. Direct interpolation +Denoted by {{code here}}. Interpolated as direct code replace. The value/output of the variable, function call, or the expression must not include spaces. ```python ## GOOD @@ -116,23 +116,23 @@ options = {'version': '-v'} >git status {{options['version']}} ``` -### 11. Dynamic interpolation -Denoted by {{{ any python variable, function call, or expression here }}}. The output of the variable, function call, or the expression must still not include spaces. +### 11. f-string interpolation +Denoted by f{ any python variable, function call, or expression here }. Interpolated as f-string. The output of the variable, function call, or the expression must still not include spaces. ```python ## GOOD # git -h options = {'version': '-v', 'help': '-h'} ->git {{{options['h']}}} +>git f{options['h']} # kubectl get pods --show-labels -n coffee namespace = "coffee" ->kubectl get pods {{{"--" + "-".join(['show', 'labels'])}}} -n {{{namespace}}} +>kubectl get pods f{"--" + "-".join(['show', 'labels'])} -n f{namespace} ## BAD option = "-s -v" ->git status {{{ option }}} +>git status f{option} ``` #### Also works inside methods! diff --git a/pybash.py b/pybash.py index fc9d593..e8129d8 100644 --- a/pybash.py +++ b/pybash.py @@ -1,5 +1,6 @@ import re import shlex +from typing import Callable, Union import token_utils from ideas import import_hook @@ -7,16 +8,16 @@ def source_init(): """Adds subprocess import""" - import_subprocess = "import subprocess" - return import_subprocess + return "import subprocess" def add_hook(**_kwargs): """Creates and automatically adds the import hook in sys.meta_path""" - hook = import_hook.create_hook( - hook_name=__name__, transform_source=Transformer.transform_source, source_init=source_init + return import_hook.create_hook( + hook_name=__name__, + transform_source=Transformer.transform_source, + source_init=source_init, ) - return hook class InvalidInterpolation(Exception): @@ -26,26 +27,26 @@ class InvalidInterpolation(Exception): class Processor: command_char = ">" - def __init__(self, token: str) -> None: + def __init__(self, token: token_utils.Token) -> None: self.token = token self.token_line = token.line self.parse() - def parse(self): + def parse(self) -> None: self.parsed_line = shlex.split(self.token_line) self.command = Commander.get_bash_command(self.parsed_line, command_char=self.command_char) def transform(self) -> token_utils.Token: raise NotImplementedError - def interpolate(self) -> str: - self.token.string = Processor.dynamic_interpolate(self.token_line, self.token.string) - self.token.string = Processor.static_interpolate(self.token.string) + def interpolate(self) -> None: + self.token.string = Processor.fstring_interpolate(self.token_line, self.token.string) + self.token.string = Processor.direct_interpolate(self.token.string) @staticmethod - def dynamic_interpolate(token_string: str, parsed_command: str) -> str: - """Process {{{ dynamic interpolations }}} and substitute. - Dynamic interpolations are denotated by a {{{ }}} with any expression inside. + def fstring_interpolate(token_string: str, parsed_command: str) -> str: + """Process f{ dynamic interpolations } and substitute. + Dynamic interpolations are denotated by a f{ } with any expression inside. Substitution in the parsed command string happens relative to the order of the interpolations in the original command string. Args: @@ -55,19 +56,20 @@ def dynamic_interpolate(token_string: str, parsed_command: str) -> str: Returns: str: Interpolated parsed command string """ - pattern = r'{{{(.+?)}}}' + pattern = r'f{(.+?)}' subs = re.findall(pattern, token_string) if not subs: return parsed_command + command_pattern = f'\"{pattern}\"' for sub in subs: - parsed_command = re.sub(pattern, '" + ' + sub + ' + "', parsed_command, 1) + parsed_command = re.sub(command_pattern, 'f\"\"\"{' + sub + '}\"\"\"', parsed_command, 1) return parsed_command @staticmethod - def static_interpolate(string: str) -> str: + def direct_interpolate(string: str) -> str: """Process {{ static interpolations }} and substitute. Static interpolations are denotated by a {{ }} with a variable or a function call inside. Substitution happens directly on the parsed command string. Therefore, certain characters cannot be interpolated as they get parsed out before substitution. @@ -112,12 +114,12 @@ def transform(self) -> token_utils.Token: class Variablized(Processor): # a = >cat test.txt - def parse(self): + def parse(self) -> None: self.parsed_line = shlex.split(self.token.line) self.start_index = Commander.get_start_index(self.parsed_line) self.command = Commander.get_bash_command(self.parsed_line, start_index=self.start_index) - def transform(self): + def transform(self) -> None: pipeline_command = Pipeline(self.command).parse_command(variablized=True) if pipeline_command != self.command: self.token.string = pipeline_command @@ -130,13 +132,13 @@ def transform(self): class Wrapped(Processor): # print(>cat test.txt) - def parse(self): + def parse(self) -> None: self.parsed_line = shlex.split(self.token.line) self.raw_line = [tok for tok in self.token.line.split(' ') if tok] self.start_index = Commander.get_start_index(self.parsed_line) self.command = Commander.get_bash_command(self.parsed_line, start_index=self.start_index, wrapped=True) - def transform(self): + def transform(self) -> token_utils.Token: # shlex strips out single quotes and double quotes-- use raw_line for the code around the wrapped command self.token.string = ( ' '.join(self.raw_line[: self.start_index]) @@ -165,20 +167,16 @@ def transform_source(source, **_kwargs): new_tokens.extend(line) continue - # matches exact token - token_match = [tokenizer for match, tokenizer in Transformer.tokenizers.items() if token == match] - if token_match: + if token_match := [tokenizer for match, tokenizer in Transformer.tokenizers.items() if token == match]: parser = token_match[0](token) parser.transform() parser.interpolate() new_tokens.append(parser.token) continue - # matches anywhere in line - greedy_match = [ + if greedy_match := [ tokenizer for match, tokenizer in Transformer.greedy_tokenizers.items() if match in token.line - ] - if greedy_match: + ]: parser = greedy_match[0](token) parser.transform() parser.interpolate() @@ -197,7 +195,7 @@ class Pipers: OPS = ['|', '>', '>>', '<', '&&'] @classmethod - def get_piper(cls, op: str): + def get_piper(cls, op: str) -> Callable: if op == '|': # Pipe output to next command return cls.chain_pipe_command @@ -213,19 +211,20 @@ def get_piper(cls, op: str): elif op == '&&': # Run next command only if previous succeeds return cls.chain_and_command - return None + + raise NotImplementedError @classmethod def chain_iredirect_command( cls, command: list, pipeline: list, start_index: int = 0, fmode: str = "r", fvar: str = "fout", **kwargs - ): + ) -> str: first_idx, _ = pipeline.pop(0) pre_command = command[start_index:first_idx] filename = command[first_idx + 1 : first_idx + 2][0] fout = f'open("{filename}", "{fmode}")' - if len(pipeline) == 0: + if not pipeline: # out to file cmd1 = Commander.build_subprocess_list_cmd("run", pre_command, stdin=fvar, **kwargs) return f"{fvar} = {fout}; cmd1 = {cmd1}" @@ -233,9 +232,10 @@ def chain_iredirect_command( cmd1 = Commander.build_subprocess_list_cmd("Popen", pre_command, stdin=fvar, stdout="subprocess.PIPE", **kwargs) out = f"{fvar} = {fout}; cmd1 = {cmd1};" - while len(pipeline) > 0: + while pipeline: idx, piper = pipeline[0] fvar = f"fout{idx}" + cmd = "" if piper == '>': # >sort < test.txt > test2.txt cmd = cls.write_to_file( @@ -259,37 +259,36 @@ def chain_iredirect_command( @classmethod def write_to_file( cls, command: list, pipeline: list, reader: str, start_index: int = 0, fvar: str = 'fout', fmode: str = 'wb' - ): + ) -> str: first_idx, _ = pipeline.pop(0) filename = command[first_idx + 1 : first_idx + 2][0] - cmd = f'{fvar} = open("{filename}", "{fmode}"); {fvar}.write({reader});' - return cmd + return f'{fvar} = open("{filename}", "{fmode}"); {fvar}.write({reader});' @classmethod def chain_and_command(cls, command: list, pipeline: list, **kwargs): raise NotImplementedError @classmethod - def chain_pipe_command(cls, command: list, pipeline: list, start_index: int = 0, chained: bool = False, **kwargs): + def chain_pipe_command( + cls, command: list, pipeline: list, start_index: int = 0, chained: bool = False, **kwargs + ) -> str: first_idx, _ = pipeline.pop(0) pre_command = command[start_index:first_idx] - - if not chained: - cmd1 = Commander.build_subprocess_list_cmd('Popen', pre_command, stdout='subprocess.PIPE', **kwargs) - - if len(pipeline) == 0: + cmd1 = ( + "" + if chained + else Commander.build_subprocess_list_cmd('Popen', pre_command, stdout='subprocess.PIPE', **kwargs) + ) + cmd2 = "" + if not pipeline: ## No other pipes post_command = command[first_idx + 1 :] cmd2 = Commander.build_subprocess_list_cmd('run', post_command, stdin='cmd1.stdout') - if not chained: - return f"cmd1 = {cmd1}; cmd2 = {cmd2}" - else: - return f"cmd2 = {cmd2}" - - out = f"cmd1 = {cmd1};" if not chained else "" - while len(pipeline) > 0: + return f"cmd2 = {cmd2}" if chained else f"cmd1 = {cmd1}; cmd2 = {cmd2}" + out = "" if chained else f"cmd1 = {cmd1};" + while pipeline: idx, piper = pipeline[0] cmd = cls.get_piper(piper)(command, pipeline, start_index=first_idx + 1, stdin="cmd1.stdout") out += cmd @@ -307,7 +306,7 @@ def chain_redirect( fmode: str = "wb", chained: bool = False, **kwargs, - ): + ) -> str: first_idx, _ = pipeline.pop(0) pre_command = command[start_index:first_idx] filename = command[first_idx + 1 : first_idx + 2][0] @@ -320,11 +319,11 @@ def chain_redirect( fout = f'open("{filename}", "{fmode}")' cmd1 = Commander.build_subprocess_list_cmd("run", pre_command, stdout=fvar, **kwargs) - if len(pipeline) == 0: + if not pipeline: return f"{fvar} = {fout}; cmd1 = {cmd1}" out = f"{fvar} = {fout}; cmd1 = {cmd1};" - while len(pipeline) > 0: + while pipeline: idx, piper = pipeline[0] fvar = f"fout{idx}" if piper in ['>', '>>']: @@ -339,13 +338,13 @@ def chain_redirect( @classmethod def chain_sredirect_command( cls, command: list, pipeline: list, start_index: int = 0, fvar: str = "fout", chained: bool = False, **kwargs - ): + ) -> str: return cls.chain_redirect(command, pipeline, start_index, fmode="wb", fvar=fvar, chained=chained, **kwargs) @classmethod def chain_dredirect_command( cls, command: list, pipeline: list, start_index: int = 0, fvar: str = "fout", chained: bool = False, **kwargs - ): + ) -> str: return cls.chain_redirect(command, pipeline, start_index, fmode="ab", fvar=fvar, chained=chained, **kwargs) @@ -354,11 +353,11 @@ class Pipeline: __slots__ = ['command', 'pipeline'] - def __init__(self, command: list): + def __init__(self, command: list[str]): self.command = command self.pipeline = [(i, arg) for i, arg in enumerate(self.command) if arg in Pipers.OPS] - def parse_command(self, variablized: bool = False, **kwargs): + def parse_command(self, variablized: bool = False, **kwargs) -> Union[list[str], str]: if not self.pipeline: return self.command @@ -383,9 +382,14 @@ def get_start_index(parsed_line: list) -> int: if '>' in val: return i + return 0 + @staticmethod def get_bash_command( - parsed_line: list, start_index: int = None, wrapped: bool = None, command_char: str = ">" + parsed_line: list, + start_index: Union[int, None] = None, + wrapped: Union[bool, None] = None, + command_char: str = ">", ) -> list: """Parses line to bash command diff --git a/pyproject.toml b/pyproject.toml index e4790e8..5369c29 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "PyBash" -version = "0.2.3" +version = "0.2.4" description = ">execute bash commands from python easily" authors = ["Jay"] readme = "README.md" diff --git a/test_pybash.py b/test_pybash.py index 4621b50..7550b98 100644 --- a/test_pybash.py +++ b/test_pybash.py @@ -97,7 +97,7 @@ def test_shell_commands(): assert run_bash("$ls .github/*") == 'subprocess.run("ls .github/*", shell=True)\n' -def test_static_interpolate(): +def test_direct_interpolate(): assert run_bash(">git {{command}} {{option}}") == 'subprocess.run(["git","" + command + "","" + option + ""])\n' assert ( run_bash(">git {{command}} {{process(option)}}") @@ -109,12 +109,12 @@ def test_static_interpolate(): ) -def test_dynamic_interpolate(): +def test_fstring_interpolate(): assert ( - run_bash(">kubectl get pods {{{\"--\" + \"-\".join(['show', 'labels'])}}} -n {{{ namespace }}}") - == 'subprocess.run(["kubectl","get","pods","" + "--" + "-".join([\'show\', \'labels\']) + "","-n","" + namespace + ""])\n' + run_bash(">kubectl get pods f{\"--\" + \"-\".join(['show', 'labels'])} -n f{ namespace }") + == 'subprocess.run(["kubectl","get","pods",f"""{"--" + "-".join([\'show\', \'labels\'])}""","-n",f"""{ namespace }"""])\n' ) - assert run_bash(">git {{{options['h']}}}") == 'subprocess.run(["git","" + options[\'h\'] + ""])\n' + assert run_bash(">git f{options['h']}") == 'subprocess.run(["git",f"""{options[\'h\']}"""])\n' def test_invalid_interpolate():