Created
January 10, 2025 19:45
-
-
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
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/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