Skip to content

Instantly share code, notes, and snippets.

@SkYNewZ
Last active November 2, 2020 19:57
Show Gist options
  • Save SkYNewZ/b5dcb847322d9903afcb8363a60c2a3b to your computer and use it in GitHub Desktop.
Save SkYNewZ/b5dcb847322d9903afcb8363a60c2a3b to your computer and use it in GitHub Desktop.
Python37 Google Cloud Logger with urllib3 handler
"""
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
@Hausfrau17
Copy link

``

@SkYNewZ
Copy link
Author

SkYNewZ commented Nov 2, 2020

``

Yes ?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment