Three basic elements of logging in Python:
Define the three elements in the same order correct order:
FORMAT is composed in Python % format, which is an old way to describe strings.
Most elements are described as %(name)s, where the final s character means string format.
Formatter Args:
asctime sets the timestamp in a human-readable format.
datefmt argument to follow the ISO 8601 format.ISO 8601 form.%(msecs)d with a d at the end means that we print the value as an integer.
name is the name of the logger, as we will describe later.
APP to differentiate between different applications.levelname is the severity of the log, such as INFO, WARNING, or ERRORmessage, finally, is the log message.FORMAT = '%(asctime)s.%(msecs)dZ:APP:%(name)s:%(levelname)s:%(message)s'
formatter = logging.Formatter(FORMAT, datefmt="%Y-%m-%dT%H:%M:%S")
handler = logging.StreamHandler(stream=sys.stdout)
handler.setFormatter(formatter)
The handler is a StreamHandler, and we set the destination of the stream to be sys.stdout.
This is the Python-defined variable that points to stdout.
Once the handler is defined, we can create the logger.
Basic Example:
import sys
import logging
# Define the format
FORMAT = '%(asctime)s.%(msecs)dZ:APP:%(name)s:%(levelname)s:%(message)s'
formatter = logging.Formatter(FORMAT, datefmt="%Y-%m-%dT%H:%M:%S")
# Create a handler that sends the logs to stdout
handler = logging.StreamHandler(stream=sys.stdout)
handler.setFormatter(formatter)
# Create a logger with name 'mylogger', adding the handler and setting the level to INFO
logger = logging.getLogger('mylogger')
logger.addHandler(handler)
logger.setLevel(logging.INFO)
# Generate three logs
logger.warning('This is a warning message')
logger.info('This is an info message')
logger.debug('This is a debug message, not to be displayed')
import logging.config
logging.config.dictConfig({
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"default": {
"format": "%(asctime)s [PID %(process)d] [Thread %(thread)d] [%(levelname)s] [%(name)s] %(message)s"
}
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"level": "INFO",
"formatter": "default",
"stream": "ext://sys.stdout"
}
},
"root": {
"level": "INFO",
"handlers": [
"console"
]
}
})
LOGGER = logging.getLogger()
LOGGER.info('logged')
import json
import requests
import logging
import os
from logging.config import dictConfig
from flask import Flask, make_response, request
# debug settings
debug = eval(os.environ.get("DEBUG", "False"))
# for sending error logs to slack
class HTTPSlackHandler(logging.Handler):
def emit(self, record):
log_entry = self.format(record)
json_text = json.dumps({"text": log_entry})
url = 'https://hooks.slack.com/services/<org_id>/<api_key>'
return requests.post(url, json_text, headers={"Content-type": "application/json"}).content
dictConfig({
"version": 1,
"disable_existing_loggers": True,
"formatters": {
"default": {
"format": "[%(asctime)s] %(levelname)s in %(module)s: %(message)s",
},
"access": {
"format": "%(message)s",
}
},
"handlers": {
"console": {
"level": "INFO",
"class": "logging.StreamHandler",
"formatter": "default",
"stream": "ext://sys.stdout",
},
"email": {
"class": "logging.handlers.SMTPHandler",
"formatter": "default",
"level": "ERROR",
"mailhost": ("smtp.example.com", 587),
"fromaddr": "devops@example.com",
"toaddrs": ["receiver@example.com", "receiver2@example.com"],
"subject": "Error Logs",
"credentials": ("username", "password"),
},
"slack": {
"class": "app.HTTPSlackHandler",
"formatter": "default",
"level": "ERROR",
},
"error_file": {
"class": "logging.handlers.RotatingFileHandler",
"formatter": "default",
"filename": "/var/log/gunicorn.error.log",
"maxBytes": 10000,
"backupCount": 10,
"delay": "True",
},
"access_file": {
"class": "logging.handlers.RotatingFileHandler",
"formatter": "access",
"filename": "/var/log/gunicorn.access.log",
"maxBytes": 10000,
"backupCount": 10,
"delay": "True",
}
},
"loggers": {
"gunicorn.error": {
"handlers": ["console"] if debug else ["console", "slack", "error_file"],
"level": "INFO",
"propagate": False,
},
"gunicorn.access": {
"handlers": ["console"] if debug else ["console", "access_file"],
"level": "INFO",
"propagate": False,
}
},
"root": {
"level": "DEBUG" if debug else "INFO",
"handlers": ["console"] if debug else ["console", "slack"],
}
})
app = Flask(__name__)
@app.route("/status", methods=["GET"])
def health_check():
logging.debug("debug log")
logging.info("info log")
logging.warning("warning log")
logging.error("error log")
# logging.exception("exception log")
return make_response("OK", 200)
if __name__ == "__main__":
app.run(debug=debug, host="0.0.0.0", port="5000")
# Run with Gunicorn
# gunicorn app:app -b 0.0.0.0:5000 --workers 2 -k gevent --timeout 300 --worker-connections 1000 --max-requests 1000000 --limit-request-line 8190 --access-logfile '-' --error-logfile '-'
import logging
from fastapi.logger import logger
from os import environ
LOG_ENV: str = environ.get('WHICH_LOGGER', 'uvicorn')
def api_logger():
# filename = f"{os.getcwd()}/logs.log"
logging.basicConfig(
format='%(asctime)s-%(process)d-%(levelname)s-%(funcName)s-%(message)s',
datefmt='%d-%b-%y %H:%M:%S',
level=logging.DEBUG)
get_logger = logging.getLogger(LOG_ENV)
logger.handlers = get_logger.handlers
logger.setLevel(get_logger.level)
return logger
logger = api_logger()
# https://github.com/tiangolo/uvicorn-gunicorn-fastapi-docker/issues/19#issuecomment-620810957
def api_logger():
from fastapi.logger import logger
get_logger = logging.getLogger(settings.WHICH_LOGGER)
logger.handlers = get_logger.handlers
logger.setLevel(get_logger.level)
return logger
logger = api_logger()
import logging
from fastapi.logger import logger
from loguru import logger
# https://github.com/tiangolo/uvicorn-gunicorn-fastapi-docker/issues/19#issuecomment-620810957
get_logger = logging.getLogger('uvicorn')
# or
# get_logger = logging.getLogger('gunicorn.error')
logger.handlers = get_logger.handlers
logger.setLevel(get_logger.level)
logger.add('uvicorn', format="{time} {level} {message}", level="INFO")
logger.add('uvicorn', format="{time} {level} {message}", level="INFO")
def setup_logger():
# Prints logger info to terminal
logger = logging.getLogger()
logger.setLevel(logging.INFO) # Change this to DEBUG if you want a lot more info
ch = logging.StreamHandler()
# create formatter
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
ch.setFormatter(formatter)
logger.addHandler(ch)
return logger
logger.add('gunicorn', format="{time} {level} {message}", level="INFO", enqueue=True)
logger.add('uvicorn', format="{time} {level} {message}", level="INFO")
logger.add('uvicorn', colorize=True, format="<green>{time}</green> <red>{message}</red>", enqueue=True)
logger.add('gunicorn', format="{time} {level} {message}", level="INFO", enqueue=True)
import os
import sys
import time
import boto3
import watchtower
import logging.handlers
def fast_logging():
try:
boto3_session = boto3.session.Session()
LOG_GROUP = "Fast-Logging"
log_level = DEBUG
stream_name = os.path.join("fast_logging"+ "_" + time.strftime("%Y%m%d") + '.log')
logger = logging.getLogger("Fast-Logger")
logger.setLevel(log_level)
formatter = logging.Formatter(
'%(name)s - %(levelname)s - %(filename)s - %(module)s: %(funcName)s: '
'%(lineno)d - %(message)s'
)
log_handler = watchtower.CloudWatchLogHandler(
log_group=LOG_GROUP,
boto3_session=boto3_session,
stream_name=stream_name
)
log_handler.setFormatter(formatter)
logger.addHandler(log_handler)
return logger
except Exception as e:
raise e
logger = fast_logging()
import logging
import scrapy
from scrapy.crawler import CrawlerProcess
from scrapy.linkextractors.lxmlhtml import LxmlLinkExtractor
logging.getLogger('scrapy').propagate = False
import logging
from scrapy import signals
from scrapy.utils.log import (
LogCounterHandler,
configure_logging,
log_scrapy_info,
get_scrapy_root_handler,
install_scrapy_root_handler
)
logger = logging.getLogger(__name__)