Snippet - Check connection to the url in Python

  1# #############################################################################
  2# Simple script to check a connection to url without any third-level packages.
  3# The script can be using as a standalone script or as a lambda function.
  4#
  5# Usage:
  6#   python check_connection.py <url1> <url2> ... <urlN>
  7#
  8# Usage as a lambda:
  9#   1. Runtime: python 3.8+
 10#   2. Memory: minimum
 11#   3. Copy this file to the lambda source directory
 12#   4. Use "check_connection.lambda_handler" method name as a lambda handler
 13#   5. You can use a payload with urls to run the lambda:
 14#       {"urls": ["url1", "url2", "url3",...]}
 15#
 16# #############################################################################
 17from __future__ import annotations
 18import json
 19import os
 20import sys
 21import logging as pylogging
 22import urllib.parse
 23import urllib.request
 24from typing import Optional, Union
 25
 26
 27def get_logger(name: Optional[str] = None, level: Union[str, None] = pylogging.INFO,
 28               message_format: Optional[str] = None):
 29    logger = pylogging.getLogger(name)
 30    logger.setLevel(pylogging.DEBUG)
 31    if os.environ.get("AWS_LAMBDA_FUNCTION_NAME"):
 32        message_format = message_format or "[%(asctime)s] {%(aws_request_id)s} %(levelname)s %(message)s\n"
 33        date_format = "%H:%M:%S"
 34        formatter = pylogging.Formatter(message_format, datefmt=date_format)
 35        logger.handlers[0].setFormatter(formatter)
 36    else:
 37        message_format = message_format or "%(message)s"
 38        date_format = None
 39    if not logger.handlers:
 40        stream_handler = pylogging.StreamHandler()
 41        formatter = pylogging.Formatter(message_format, datefmt=date_format)
 42        stream_handler.setFormatter(formatter)
 43        stream_handler.setLevel(level)
 44        logger.addHandler(stream_handler)
 45    return logger
 46
 47def make_request(logger, method: str, url: str) -> dict:
 48    status_code = 0
 49    content = ""
 50    logger.info(f"Make request to {url}")
 51    try:
 52        req = urllib.request.Request(url, method=method)
 53        logger.info(f"Request: {req}")
 54        resp = urllib.request.urlopen(req, timeout=5)
 55        status_code = resp.getcode()
 56        logger.info(f"Response: [{status_code}]")
 57        content = resp.read() or ""
 58        if isinstance(content, (bytearray, bytes)):
 59            content = content.decode("utf-8")
 60        content = content.strip()
 61        logger.info("Content:")
 62        logger.info(content[:200])
 63    except urllib.error.HTTPError as ex:
 64        content = str(ex)
 65        status_code = int(ex.code)
 66        logger.error(f"Response exception [{ex}]")
 67    except urllib.error.URLError as ex:
 68        content = str(ex)
 69        status_code = 400
 70        logger.error(f"Response exception [{ex}]")
 71    return {
 72        "url": url,
 73        "response": {
 74            "status": status_code,
 75            "content": content[:200],
 76        },
 77    }
 78
 79def lambda_handler(event: dict, _):
 80    event = event or {}
 81    logger = get_logger()
 82    logger.info(f"START lambda: event={event}")
 83    urls = event.get("urls") or event.get("url") or os.environ.get("URLS") \
 84        or os.environ.get("URL") or "https://github.com"
 85    if not isinstance(urls, list):
 86        urls = [s for s in [s.strip() for s in str(urls).split(",")] if s]
 87    result = []
 88    if urls:
 89        logger.info(f"Check connection to the urls: {urls}")
 90        logger.info("-" * 40)
 91        for url in urls:
 92            target_url = f"http://{url.lstrip('/')}" if not url.startswith("http") else url
 93            logger.debug(f"Connect to {target_url}")
 94            resp = make_request(logger, "GET", str(target_url))
 95            result.append(resp)
 96        logger.info("-" * 40)
 97    else:
 98        result.append({"response": {"status": 400, "content": "Urls list is empty"}})
 99
100    main_status = max([r["response"]["status"] for r in result]) or 500
101    logger.info(f"Status={main_status}")
102    logger.info(f"END lambda: event={event}")
103    lambda_response = {
104        "status": main_status,
105        "message": "OK" if 200 <= main_status < 300 else "ERROR",
106        "data": result,
107    }
108    logger.info(json.dumps(lambda_response, indent=2))
109    return lambda_response
110
111if __name__ == "__main__":
112    lambda_handler({"urls": sys.argv[1:]}, {})