Skip to content

misc

Miscellaneous utility functions for BitBazaar.

StdCapture(stderr=False)

Bases: list

Capture stdout/stderr for the duration of a with block.

(e.g. print statements)

Example:

with StdCapture(stderr=True) as out: # By default only captures stdout
    print('hello')
    sys.stderr.write('world')
print(out)  # ['hello', 'world']

Creation of new capturer.

By default only stdout is captured, stderr=True enables capturing stderr too.

Source code in py/bitbazaar/misc/_std_capture.py
def __init__(self, stderr: bool = False):
    """Creation of new capturer.

    By default only stdout is captured, stderr=True enables capturing stderr too.
    """
    # Prep all instance vars:
    self.stderr_capture = stderr
    self._out = []
    self._stdout = sys.stdout
    self._stderr = sys.stderr
    self._buf = io.StringIO()

__enter__()

Entering the capturing context.

Source code in py/bitbazaar/misc/_std_capture.py
def __enter__(self) -> list[str]:
    """Entering the capturing context."""
    # Overwrite sinks which are being captured with the buffer:
    sys.stdout = self._buf
    if self.stderr_capture:
        sys.stderr = self._buf

    return self._out

__exit__(*args)

On context exit.

Source code in py/bitbazaar/misc/_std_capture.py
def __exit__(self, *args):  # type: ignore
    """On context exit."""
    # First reset the global streams:
    sys.stdout = self._stdout
    sys.stderr = self._stderr

    self._out.extend(self._buf.getvalue().splitlines())

copy_sig(f)

Keep e.g. a class's init signature when subclassing.

From: https://github.com/python/typing/issues/769#issuecomment-903760354

Source code in py/bitbazaar/misc/__init__.py
def copy_sig(f: _T) -> tp.Callable[[tp.Any], _T]:
    """Keep e.g. a class's __init__ signature when subclassing.

    From: https://github.com/python/typing/issues/769#issuecomment-903760354
    """
    return lambda x: x

in_ci()

Returns true if it looks like the program is running from a CI service, e.g. Github Actions.

Source code in py/bitbazaar/misc/__init__.py
def in_ci() -> bool:
    """Returns true if it looks like the program is running from a CI service, e.g. Github Actions."""
    return any([var in os.environ for var in _CI_ENV_VARS])

is_tcp_port_listening(host, port)

Check if something is listening on a certain tcp port or not.

Source code in py/bitbazaar/misc/__init__.py
def is_tcp_port_listening(
    host: str, port: int
) -> bool:  # pragma: no cover (is covered but not in CI)
    """Check if something is listening on a certain tcp port or not."""
    try:
        # Create a TCP socket
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(1)  # Set timeout to 1 second

        # Attempt to establish a connection to the port
        s.connect((host, port))

        # If connection is successful, something is listening on the port
        s.close()
        return True
    except OSError:
        return False

Last update: August 7, 2024
Created: August 7, 2024