PK!C=shell_utils/__init__.py# -*- coding: utf-8 -*- """Top-level package for shell-utils.""" from shell_utils.shell_utils import * from shell_utils.notify import notify PK!d&shell_utils/cli.py""" Usage: [OPTIONS] COMMAND [ARGS]... A cli for shell-utils. Options: --help Show this message and exit. """ from pathlib import Path from shell_utils import shell import click @click.group() def cli(): """A cli for shell_utils.""" pass @cli.command() def generate_runner(): """Generate a run.py script in the current directory.""" from shell_utils import runner runner_path = Path('run.py') if runner_path.exists(): raise EnvironmentError('run.py already exists in current directory') click.secho('writing content to run.py', fg='yellow') runner_path.write_text(Path(runner.__file__).read_text()) shell('chmod +x run.py') if __name__ == "__main__": cli() PK!2f f shell_utils/notify.pyimport subprocess as sp import typing as typ import logging import shlex import sys from functools import wraps import click def notify(message: str, title=None, subtitle=None, sound=None): """ Send a Mac OS notification. Args: message: the notification body title: the title of the notification subtitle: the subtitle of the notification sound: the sound the notification makes see https://apple.stackexchange.com/questions/57412/how-can-i-trigger-a-notification-center-notification-from-an-applescript-or-shel/115373#115373 """ if title is None: title = 'heads up' if subtitle is None: subtitle = 'something happened' if sys.platform != 'darwin': logging.warning('This function is designed to work on Mac OS') # There is probably a less hacky way to escape single quotes safely # but I haven't gotten to it message = message.replace("'", '') command = f"""osascript -e 'display notification "{message}" with title "{title}" subtitle "{subtitle}" sound name "{sound}"' """ sp.run(shlex.split(command), check=False) def notice(message: typ.Optional[str] = None, title=None, subtitle=None, sound=None): """ Returns a decorator that allows you to be notified when a function returns. Args: message: the notification body title: the title of the notification subtitle: the subtitle of the notification sound: the sound the notification makes Returns: a function """ def decorator(func): """ Send notification that task has finished. Especially useful for long-running tasks """ @wraps(func) def inner(*args, **kwargs): nonlocal message, title, subtitle, sound title = getattr(func, '__name__') + ' finished' if title is None else title subtitle = 'Success!' if subtitle is None else subtitle result = None try: _result = func(*args, **kwargs) if isinstance(_result, str): result = _result except: subtitle = 'Failure' raise finally: if message is not None: pass elif result is not None: message = result else: message = '' notify(message, title=title, subtitle=subtitle, sound=sound) return inner return decorator @click.command() @click.argument('message') @click.option('--title', help='the notification title') @click.option('--subtitle', help='the notification subtitle') @click.option('--sound', help='the notification sound') def notify_command(message, title, subtitle, sound): """Notification cli tool.""" notify(message, title=title, subtitle=subtitle, sound=sound) PK!Ew  shell_utils/runner.py#!/usr/bin/env python3 import os from pathlib import Path from shell_utils import shell, cd, env, path, quiet import click @click.group() def main(): """ Development tasks; programmatically generated """ # ensure we're running commands from project root root = Path(__file__).parent.absolute() cwd = Path().absolute() if root != cwd: click.secho(f'Navigating from {cwd} to {root}', fg='yellow') os.chdir(root) if __name__ == '__main__': main() PK!55shell_utils/shell_utils.py# -*- coding: utf-8 -*- """Main module.""" import copy import os import subprocess as sp import textwrap import types import typing as T from contextlib import contextmanager from getpass import getuser from socket import gethostname import click Pathy = T.Union[os.PathLike, str] def _bool(self: sp.CompletedProcess) -> bool: """ Return True if return code is zero else false. Args: self: sp.CompletedProcess Returns: True or False """ return self.returncode == 0 def shell(command: str, check=True, capture=False, silent=False, dedent=True, strip=True, **kwargs ) -> sp.CompletedProcess: """ Run the command in a shell. !!! Make sure you trust the input to this command !!! Args: command: the command to be run check: raise exception if return code not zero capture: if set to True, captures stdout and stderr, making them available as stdout and stderr attributes on the returned CompletedProcess. This also means the command's stdout and stderr won't be piped to FD 1 and 2 by default silent: disable the printing of the command that's being run prior to execution dedent: de-dent command string; useful if it's a bash script written within a function in your module strip: strip ends of command string of newlines and whitespace prior to execution kwargs: passed to subprocess.run as-is Returns: Completed Process """ user = click.style(getuser(), fg='green') hostname = click.style(gethostname(), fg='blue') command = textwrap.dedent(command) if dedent else command command = command.strip() if strip else command if not silent: print(f'{user}@{hostname}', click.style('executing...', fg='yellow') ) print(command) print( click.style( ('-' * max(len(l) for l in command.splitlines()) if command else ''), fg='magenta' ) ) print() process = sp.run( command, check=check, shell=True, stdout=sp.PIPE if capture else None, stderr=sp.PIPE if capture else None, **kwargs ) # override bool dunder method process._bool = types.MethodType(_bool, process) process.__class__.__bool__ = process._bool if capture: # decode stderr and stdout # keep bytes as raw_{stream} process.raw_stdout: bytes = process.stdout process.raw_stderr: bytes = process.stderr process.stdout: str = process.stdout.decode() process.stderr: str = process.stderr.decode() return process @contextmanager def cd(path_: Pathy): """Change the current working directory.""" cwd = os.getcwd() os.chdir(os.path.expanduser(path_)) yield os.chdir(cwd) @contextmanager def env(**kwargs) -> T.Iterator[os._Environ]: """Set environment variables and yield new environment dict.""" original_environment = copy.deepcopy(os.environ) for key, value in kwargs.items(): os.environ[key] = value yield os.environ os.environ = original_environment @contextmanager def path(*paths: Pathy, prepend=False, expand_user=True) -> T.Iterator[ T.List[str]]: """ Add the paths to $PATH and yield the new $PATH as a list. Args: prepend: prepend paths to $PATH else append expand_user: expands home if ~ is used in path strings """ paths_list: T.List[Pathy] = list(paths) paths_str_list: T.List[str] = [] for index, _path in enumerate(paths_list): if not isinstance(_path, str): print(f'index: {index}') paths_str_list.append(_path.__fspath__()) elif expand_user: paths_str_list.append(os.path.expanduser(_path)) original_path = os.environ['PATH'].split(':') paths_str_list = paths_str_list + \ original_path if prepend else original_path + paths_str_list with env(PATH=':'.join(paths_str_list)): yield paths_str_list @contextmanager def quiet(): """ Suppress stdout and stderr. https://stackoverflow.com/questions/11130156/suppress-stdout-stderr-print-from-python-functions """ # open null file descriptors null_file_descriptors = ( os.open(os.devnull, os.O_RDWR), os.open(os.devnull, os.O_RDWR) ) # save stdout and stderr stdout_and_stderr = (os.dup(1), os.dup(2)) # assign the null pointers to stdout and stderr null_fd1, null_fd2 = null_file_descriptors os.dup2(null_fd1, 1) os.dup2(null_fd2, 2) yield # re-assign the real stdout/stderr back to (1) and (2) stdout, stderr = stdout_and_stderr os.dup2(stdout, 1) os.dup2(stderr, 2) # close all file descriptors. for fd in null_file_descriptors + stdout_and_stderr: os.close(fd) # alias bash bash = shell __all__ = [ 'shell', 'bash', 'cd', 'env', 'path', 'quiet', ] PK!Hg@\*shell_utils-2.0.dist-info/entry_points.txtN+I/N.,()/L-Hɉ/-)փYAļ.$(s2 PK! J*//!shell_utils-2.0.dist-info/LICENSECopyright 2018 Stephan Fitzpatrick Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.PK!HMWXshell_utils-2.0.dist-info/WHEEL A н#Z@Z|Jl~6蓅 Λ MU4[PYBpYD*Mͯ#ڪ/̚?ݭ'%nPK!H&"shell_utils-2.0.dist-info/METADATAXo8НmI[ DQYъo瑼vxa}އ}+(ߥJ3"{މJθʲL[J_붪5Ip:] m.턗Va:i*l^JֹΦӍrv庚zg|:O/U.k /үKZ[Q8lEo^*+?6jCj$D'|6?4+ 61Tᗢ޴0OW{S峳'j||5yƷO8cn6Px^|gWcv!mnTC@K׵Ko `7$o {l%_z8-hJ EŬVHp [fӀWv0[Ϣ:;Ғ:~8 Gw׾9Mǭcڧrxun-p ?qwQ/$/}0JbEl E\R 1c N.t`Bg>3$:λ{gz='r lRRa4Qp᝕4ީ1W gjaP0) !;#*΃9C|ːЪov\vȅ H" Z(Ld!$\6O"5q;֣80uc<di\Qt]F œxo!F 6p" ~z/H)׼j>5pԏFk1zAI))QCpHﯝ`?.w?ڃ%'JM|d=_^-bizq7\14\D rvJҨѧߑ}4hu8$}-zb)(YGJ9C^ƝCNS=80)yOժ6hgi[jexOo?(0 a 7!sC[tõ^x_PPK!Ho shell_utils-2.0.dist-info/RECORD}л0~? Cq P.*ȥa$rħ?3حgZe#eH}%<{ޏ~*~y'Y{*쇼qEo"O*[B¢.Dj{n+dU]-DCWJE0/Xa v)eZ'w &5l8Q*qotX lhl}yc8I7qItRޫힹ^zK51 uz'P3$a/N3"G,]dBGJjST<9Kv%DE#NI+T7r%4=/ޙfULղlj!Hn 9'I82 cb&O?<&{Z`'C]gDL}"*Ue8/8#ۉٜT:`^4p8 74osPK!C=shell_utils/__init__.pyPK!d&shell_utils/cli.pyPK!2f f shell_utils/notify.pyPK!Ew  cshell_utils/runner.pyPK!55shell_utils/shell_utils.pyPK!Hg@\* &shell_utils-2.0.dist-info/entry_points.txtPK! J*//!&shell_utils-2.0.dist-info/LICENSEPK!HMWX)shell_utils-2.0.dist-info/WHEELPK!H&")shell_utils-2.0.dist-info/METADATAPK!Ho 2shell_utils-2.0.dist-info/RECORDPK 4