scwrypts/py/lib/scwrypts/io/combined_io_stream.py
yage a739d3b5a2 v4.0.0
=====================================================================

Big day! V4 is finally live. This INCLUDES some BREAKING CHANGES to ZSH
TYPE scwrypts! Please refer to the readme for upgrade details
                     (more specifically docs/upgrade/v3-to-v4.md)

Upgrade is SUPER EASY, so please take the time to do so.

--- New Features ----------------------------------------------------

- zsh type scwrypts have an upgraded runstring to improve context setup
  and simplicity to the scwrypt-writer

- scwrypts now publishes the package (scwrypts) to PyPi; this provides a
  simple way to invoke scwrypts from python-based environments as well
  as the entire scwrypts python library suite

  pip install scwrypts

- scwrypts now publishes the package (scwrypts) to npm; this provides a
  simple way to invoke scwrypts from nodesjs environments

  npm install scwrypts

--- Bug Fixes -------------------------------------------------------

- scwrypts runner prompts which use the zshbuiltin "read" now
  appropriately read input from tty, pipe, files, and user input

- virtualenv refresh now loads and prepares the scwrypts virtual
  environments correctly

--- Changes ---------------------------------------------------------

- created the (-v, --log-level) scwrypts arguments as improvements of
  and replacements to the --verbose and --no-log flags
     - (-n) is now an alias for (--log-level 0)
     - (--no-log) is the same as (-n) for compatibility, but will be removed in 4.2

- zsh/lib/utils/io print functions now *interact with log-level* various
  log levels will now only display the appropriate console prints for
  the specified log level

- zsh/lib/utils/io:INFO has been renamed to DEBUG to align with
  log-level output; please use DEBUG for debug messages and REMINDER for
  important user messages

- created zsh/lib/utils/io:FZF_USER_INPUT as a *drop-in replacement* for
  the confusing FZF_HEAD and FZF_TAIL commands. Update by literally
  changing any instances of FZF_HEAD or FZF_TAIL with FZF_USER_INPUT
     - FZF_HEAD and FZF_TAIL will be removed in 4.2

- zsh/lib/utils/io:READ (and other zshbuiltin/read-based prompts) now
  accept a --force-user-input flag in case important checks should
  require an admin's approval. This flag will ensure that piped input
  and the `scwrypts -y` flag are ignored for the single prompt.

- zsh/lib/utils/color has been updated to use color names which match
  the ANSI color names

- zsh/hello-world has been reduced to a minimal example; this is to
  emphasize ease-of-use with v4

- zsh/sanity-check is a scwrypts/run testing helper and detailed
  starting reference (helpful since hello-world is now minimal)

- various refactor, updates, and improvements to the scwrypts runner

- migrated all zsh scwrypts and plugins to use v4 runner syntax
     - zsh
     - plugins/kubectl
     - plugins/ci

- refactored py/lib into py/lib/scwrypts (PyPi)
2024-02-20 23:51:32 -07:00

101 lines
3.0 KiB
Python

from contextlib import contextmanager
from pathlib import Path
from sys import stdin, stdout, stderr
from scwrypts.env import getenv
@contextmanager
def get_combined_stream(input_file=None, output_file=None):
'''
context manager to open an "input_file" and "output_file"
But the "files" can be pipe-streams, stdin/stdout, or even
actual files! Helpful when trying to write CLI scwrypts
which would like to accept all kinds of input and output
configurations.
'''
with get_stream(input_file, 'r') as input_stream, get_stream(output_file, 'w+') as output_stream:
yield CombinedStream(input_stream, output_stream)
def add_io_arguments(parser, allow_input=True, allow_output=True):
'''
slap these puppies onto your argparse.ArgumentParser to
allow easy use of the get_combined_stream at the command line
'''
if allow_input:
parser.add_argument(
'-i', '--input-file',
dest = 'input_file',
default = None,
help = 'path to input file; omit for stdin',
required = False,
)
if allow_output:
parser.add_argument(
'-o', '--output-file',
dest = 'output_file',
default = None,
help = 'path to output file; omit for stdout',
required = False,
)
#####################################################################
@contextmanager
def get_stream(filename=None, mode='r', encoding='utf-8', verbose=False, **kwargs):
allowed_modes = {'r', 'w', 'w+'}
if mode not in allowed_modes:
raise ValueError(f'mode "{mode}" not supported modes (must be one of {allowed_modes})')
is_read = mode == 'r'
if filename is not None:
if verbose:
print(f'opening file {filename} for {"read" if is_read else "write"}', file=stderr)
if filename[0] not in {'/', '~'}:
filename = Path(f'{getenv("EXECUTION_DIR")}/{filename}').resolve()
with open(filename, mode=mode, encoding=encoding, **kwargs) as stream:
yield stream
else:
if verbose:
print('using stdin for read' if is_read else 'using stdout for write', file=stderr)
yield stdin if is_read else stdout
if not is_read:
stdout.flush()
class CombinedStream:
def __init__(self, input_stream, output_stream):
self.input = input_stream
self.output = output_stream
def read(self, *args, **kwargs):
return self.input.read(*args, **kwargs)
def readline(self, *args, **kwargs):
return self.input.readline(*args, **kwargs)
def readlines(self, *args, **kwargs):
return self.input.readlines(*args, **kwargs)
def write(self, *args, **kwargs):
return self.output.write(*args, **kwargs)
def writeline(self, line):
x = self.output.write(f'{line}\n')
self.output.flush()
return x
def writelines(self, *args, **kwargs):
return self.output.writelines(*args, **kwargs)