Skip to content

Instantly share code, notes, and snippets.

@shollingsworth
Created January 10, 2025 19:45
Show Gist options
  • Save shollingsworth/2cb51b7ce0744aabed393353c83c5f48 to your computer and use it in GitHub Desktop.
Save shollingsworth/2cb51b7ce0744aabed393353c83c5f48 to your computer and use it in GitHub Desktop.
native python tcp socket checker similar to the bash wait-for-it project. https://github.com/vishnubob/wait-for-it
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Wait for a tcp network port to become open."""
import argparse
import socket
import sys
import time
from argparse import Namespace
SLEEP_SECONDS = 1
def wait_for_it(args: Namespace):
"""Waits for a host:port to become available."""
has_timeout = args.timeout != 0
if not has_timeout:
print(f"Running forever, no timeout configured: '{args.host}:{args.port}'")
start_time = time.time()
while True:
try:
if args.verbose:
print(f"Trying connection to {args.host}:{args.port}")
with socket.create_connection((args.host, args.port), timeout=1):
if args.verbose:
print(f"Success! {args.host}:{args.port}")
break # Connection successful
except OSError as e:
is_timedout = time.time() - start_time > args.timeout
if has_timeout and is_timedout:
raise TimeoutError(
f"Timeout waiting for {args.host}:{args.port}"
) from e
if args.verbose:
print(
f"Connect timed out: {args.host}:{args.port}, sleeping {SLEEP_SECONDS}"
)
time.sleep(SLEEP_SECONDS)
def main():
"""Run main function."""
parser = argparse.ArgumentParser(
formatter_class=argparse.RawDescriptionHelpFormatter,
description=__doc__,
)
parser.add_argument(
"--verbose",
"-v",
help="enable verbose logging",
action="store_true",
default=False,
)
parser.add_argument(
"host",
help="host / ip address",
type=str,
)
parser.add_argument(
"port",
help="tcp port number",
type=int,
)
parser.add_argument(
"--timeout",
"-t",
help="timeout",
type=int,
default=0,
)
args = parser.parse_args()
try:
wait_for_it(args)
print(f"{args.host}:{args.port} is available!")
except TimeoutError as e:
print(e)
sys.exit(1)
except KeyboardInterrupt:
print("Goodbye!")
sys.exit(0)
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment