install filter on logging level in python using dictConfig

Question

I cannot install a filter on a logging handler using dictConfig() syntax. LoggingErrorFilter.filter() is simply ignored, nothing happens.

I want to filter out error messages so that they do not appear twice in the log output. So I wrote LoggingErrorFilter class and overrode filter().

My configuration:

class LoggingErrorFilter(logging.Filter):
  def filter(self, record):
    print 'filter!'
    return record.levelno == logging.ERROR or record.levelno == logging.CRITICAL

config = {
      'version': 1,
      'disable_existing_loggers' : False,
      'formatters' : {
        'standard' : {
          'format' : '%(asctime)s %(levelname)s %(name)s::%(message)s',
        },
      },
      'handlers' : {
        'console': {
          'class' : 'logging.StreamHandler',
          'level' : level,
          'formatter' : 'standard',
          'stream' : 'ext://sys.stdout',
        },
        'errorconsole': {
          'class' : 'logging.StreamHandler',
          'level' : 'ERROR',
          'formatter' : 'standard',
          'stream' : 'ext://sys.stderr',
          'filters'  :['errorfilter',],
        },
      },
      'filters': {
        'errorfilter': {
          'class' : 'LoggingErrorFilter',
        }
      },
      'loggers' : {
        '' : {
          'handlers' : ['errorconsole','console',],
          'level' : level,
          'propagate' : True,
        },
        name : {
          'handlers' : ['errorconsole','console',],
          'level' : level,
          'propagate' : False,
        },
      },
  }
  logging.config.dictConfig(config)

What am I doing wrong here? Why is my filter ignored?

Delinquency

Answer 1

Actually, Tupteq's answer is not correct in general. The following script:

import logging
import logging.config
import sys

class MyFilter(logging.Filter):
    def __init__(self, param=None):
        self.param = param

    def filter(self, record):
        if self.param is None:
            allow = True
        else:
            allow = self.param not in record.msg
        if allow:
            record.msg = 'changed: ' + record.msg
        return allow

LOGGING = {
    'version': 1,
    'filters': {
        'myfilter': {
            '()': MyFilter,
            'param': 'noshow',
        }
    },
    'handlers': {
        'console': {
            'class': 'logging.StreamHandler',
            'filters': ['myfilter']
        }
    },
    'root': {
        'level': 'DEBUG',
        'handlers': ['console']
    },
}

if __name__ == '__main__':
    print(sys.version)
    logging.config.dictConfig(LOGGING)
    logging.debug('hello')
    logging.debug('hello - noshow')

When run, produces the following output:

$ python filtcfg.py 
2.7.5+ (default, Sep 19 2013, 13:48:49) 
[GCC 4.8.1]
changed: hello

which shows that you can configure filters using dictConfig().

Grouch

My error was that I used qoutation marks around the class name. The config dictionary expected object type, not the type's name. Thanks! – Delinquency

does it work with yml config ? I was trying but no luck – Fragment

I had to use a small factory method: def new_instance(): return MyLoggingFilter() and '()': 'MyLoggingFilter.new_instance' – Winthrop

THank you so much. My problem was that i was instantiating the class instead of passing it directly – Adala

Answer 2

You can specify a class name, but it is done with the strangely named () key, and it has to include the module name. E.g.:

 'filters': {
    'errorfilter': {
      '()' : '__main__.LoggingErrorFilter',
    }
  },

See 16.7.2.4. User-defined objects in the documentation.

Misrepresent

I don't recall how I figured this out. But the special () key is documented in docs.python.org/3/library/…. I've now added the link in the answer too. Thanks! – Misrepresent

Answer 3

if wanted to load from a json file. this is how it should be

{
  "version": 1,
  "grey": "/x1b[38;21m",
  "yellow": "/x1b[33;21m",
  "red": "/x1b[31;21m",
  "bold_red": "/x1b[31;1m",
  "reset": "/x1b[0m",
  "filters": {
    "audit_filter": {
      "()": "app.logs.AuditFilter.AuditFilter",
      "param": "audit"
    },
    "ignore_audit_filter": {
      "()": "app.logs.AuditFilter.IgnoreAuditFilter",
      "param": "audit"
    }
  },
  "formatters": {
    "package_formatter": {
      "format": "[%(asctime)s] - [%(levelname)s] - [%(name)s] : %(message)s"
    }
  },
  "handlers": {
    "console": {
      "class": "logging.StreamHandler",
      "level": "DEBUG",
      "formatter": "package_formatter"
    },
    "development_file_handler": {
      "class": "logging.FileHandler",
      "filename": "../debug.log",
      "level": "DEBUG",
      "formatter": "package_formatter",
      "filters": [
        "ignore_audit_filter"
      ]
    },
    "audit_file_handler": {
      "class": "logging.FileHandler",
      "filename": "../audit.log",
      "level": "DEBUG",
      "formatter": "package_formatter",
      "filters": [
        "audit_filter"
      ]
    }
  },
  "loggers": {
    "audit": {
      "handlers": [

        "audit_file_handler"
      ],
      "level": "INFO"
    }
  },
  "root": {
    "level": "INFO",
    "handlers": [
      "console",
      "development_file_handler"
    ]
  }
}

initialise and then attach logger and run the logger use app.logger = logging.getlogger("audit")

run the application

Fragment

Answer 4

import logging
import os
import yaml
from logging.config import dictConfig
from opencensus.ext.azure import  log_exporter
import opencensus

os.environ["APPLICATIONINSIGHTS_CONNECTION_STRING"] = '<your key>'
# you can put your logic in filters whatever you need
class FileFilter(logging.Filter):

    def __init__(self, param=None):
        self.param = param
        print("File filter initialized")

    def filter(self, record):
        # print("from file filter")
        if self.param is None:
            allow = True
            print("all msg allowed")
        else:
            if (record.msg.find("fp:") == -1):
                allow = False

            else:
                print("sending log to appinsight")
                allow = True
                temp = record.msg
                temp = temp.replace("fp: ", "")
                record.msg = temp
        return allow


class AppInsightFilter(logging.Filter):
    def __init__(self, param=None):
        self.param = param
        print("appinsight filter initialized")

    def filter(self, record):
        # print("from aap filter")
        if self.param is None:
            allow = True
            print("all msg allowed")
        else:
            if (record.msg.find("app:") == -1):
                allow = False
            else:
                allow = True
                temp = record.msg
                temp = temp.replace("app: ", "")
                record.msg = temp
        return allow

config_dict = {
        'version': 1,
        'formatters': {
            'default': {'format': '%(asctime)s | %(levelname)s | %(filename)s | %(message)s', 'datefmt': '%Y-%m-%d %H:%M:%S'}
        },
        'filters': {
            'file_filter': {
                '()': '__main__.FileFilter',
                'param': 'noshow'
            },
            'appinsight_filter': {
                '()': '__main__.AppInsightFilter',
                'param': 'noshow'
            }
            
        },
        'handlers': {
            'console': {
                'level': 'DEBUG',
                '()': 'logging.StreamHandler',
                'formatter': 'default',
                'stream': 'ext://sys.stdout'
            },
            'file': {
                'level': 'DEBUG',
                '()': 'logging.FileHandler',
                'formatter': 'default',
                'filename': 'log_path.txt',
                'filters': ['file_filter']
            },
            'appinsight': {
                'level': 'INFO',
                'filters': ['appinsight_filter'],
                '()': 'opencensus.ext.azure.log_exporter.AzureLogHandler'
                
            }
        },
        'loggers': {
            'root': {
                'level': 'DEBUG',
                'handlers': ['console', 'file', 'appinsight']
            }
        },
        'disable_existing_loggers': False
    }

dictConfig(config_dict)

logging=logging.getLogger("root")
logging.debug("Normal log debug")
logging.info("Normal log info")
logging.warning("Normal log warning")
logging.error("Normal log error")
logging.warning("fp: for app insight")
logging.warning("app: for file insight")
print("done..")

The point that I have noticed, is '()': "module.class_name" and call to getLogger is necessary.

Christcrossrow

Isn't this already mentioned in another answer? Please don't repeat answers. – Cabriole