Last active
November 2, 2020 19:57
-
-
Save SkYNewZ/b5dcb847322d9903afcb8363a60c2a3b to your computer and use it in GitHub Desktop.
Python37 Google Cloud Logger with urllib3 handler
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
| """ | |
| Used to write logs on Stackdriver logging using the https://googleapis.dev/python/logging/latest/client.html | |
| Detect if we are in Google Runtime environment and enable this logger or not | |
| Change the default format | |
| Append all urllib3 debug request into these loggers | |
| https://stackoverflow.com/questions/16337511/log-all-requests-from-the-python-requests-module | |
| https://github.com/psf/requests/issues/1297 | |
| https://stackoverflow.com/questions/11820338/replace-default-handler-of-python-logger | |
| https://stackoverflow.com/questions/879732/logging-with-filters | |
| """ | |
| import logging | |
| import os | |
| import requests | |
| import json | |
| import google.cloud.logging # Don't conflict with standard logging | |
| from google.cloud.logging.handlers import CloudLoggingHandler | |
| from google.cloud.logging.resource import Resource | |
| def setup_loging() -> logging.Logger: | |
| # Enable URL requests debug | |
| requests_log = logging.getLogger("urllib3") | |
| requests_log.setLevel(logging.DEBUG) | |
| requests_log.propagate = False | |
| # If not in GCP, use default logger | |
| if not (os.getenv("GCP_PROJECT") and os.getenv("FUNCTION_NAME")): | |
| print("Using default logger") | |
| # Create logger | |
| logger = logging.getLogger(__name__) | |
| formatter = logging.Formatter("[%(levelname)s] %(name)s - %(message)s") | |
| # Create handler | |
| ch = logging.StreamHandler() | |
| ch.setFormatter(formatter) | |
| logger.addHandler(ch) | |
| # Set default level | |
| logger.setLevel(logging.DEBUG) | |
| requests_log.addHandler(ch) | |
| return logger | |
| print("Using Google logger") | |
| client = google.cloud.logging.Client() | |
| handler = CloudLoggingHandler( | |
| client, | |
| name="cloudfunctions.googleapis.com%2Fcloud-functions", | |
| resource=Resource( | |
| type="cloud_function", | |
| labels={ | |
| "function_name": os.getenv("FUNCTION_NAME"), | |
| "region": os.getenv("FUNCTION_REGION"), | |
| }, | |
| ), | |
| ) | |
| cloud_logger = logging.getLogger("cloudLogger") | |
| cloud_logger.setLevel(logging.DEBUG) | |
| cloud_logger.addHandler(handler) | |
| requests_log.addHandler(handler) | |
| return cloud_logger | |
| logging = setup_loging() | |
| def main(request): | |
| r = requests.get("https://lemairepro.fr/not-exist") | |
| try: | |
| r.raise_for_status() | |
| return r.json() | |
| except requests.exceptions.HTTPError as e: | |
| try: | |
| m = r.json().get("Message") | |
| logging.error("Error in request, parsing JSON response") | |
| raise Exception(f"{e}: {m}") from None | |
| except json.JSONDecodeError as json_error: | |
| logging.error("json.JSONDecodeError %s: Received:%s", json_error, r.text) | |
| return "OK" | |
| if __name__ == "__main__": | |
| main(None) | |
| pass |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment