importloggingimportauxiliary_module# create logger with 'spam_application'logger=logging.getLogger('spam_application')logger.setLevel(logging.DEBUG)# create file handler which logs even debug messagesfh=logging.FileHandler('spam.log')fh.setLevel(logging.DEBUG)# create console handler with a higher log levelch=logging.StreamHandler()ch.setLevel(logging.ERROR)# create formatter and add it to the handlersformatter=logging.Formatter('%(asctime)s -%(name)s -%(levelname)s -%(message)s')fh.setFormatter(formatter)ch.setFormatter(formatter)# add the handlers to the loggerlogger.addHandler(fh)logger.addHandler(ch)logger.info('creating an instance of auxiliary_module.Auxiliary')a=auxiliary_module.Auxiliary()logger.info('created an instance of auxiliary_module.Auxiliary')logger.info('calling auxiliary_module.Auxiliary.do_something')a.do_something()logger.info('finished auxiliary_module.Auxiliary.do_something')logger.info('calling auxiliary_module.some_function()')auxiliary_module.some_function()logger.info('done with auxiliary_module.some_function()')

importlogging# create loggermodule_logger=logging.getLogger('spam_application.auxiliary')classAuxiliary:def__init__(self):self.logger=logging.getLogger('spam_application.auxiliary.Auxiliary')self.logger.info('creating an instance of Auxiliary')defdo_something(self):self.logger.info('doing something')a=1+1self.logger.info('done doing something')defsome_function():module_logger.info('received a call to "some_function"')

2005-03-23 23:47:11,663 - spam_application - INFO -   creating an instance of auxiliary_module.Auxiliary2005-03-23 23:47:11,665 - spam_application.auxiliary.Auxiliary - INFO -   creating an instance of Auxiliary2005-03-23 23:47:11,665 - spam_application - INFO -   created an instance of auxiliary_module.Auxiliary2005-03-23 23:47:11,668 - spam_application - INFO -   calling auxiliary_module.Auxiliary.do_something2005-03-23 23:47:11,668 - spam_application.auxiliary.Auxiliary - INFO -   doing something2005-03-23 23:47:11,669 - spam_application.auxiliary.Auxiliary - INFO -   done doing something2005-03-23 23:47:11,670 - spam_application - INFO -   finished auxiliary_module.Auxiliary.do_something2005-03-23 23:47:11,671 - spam_application - INFO -   calling auxiliary_module.some_function()2005-03-23 23:47:11,672 - spam_application.auxiliary - INFO -   received a call to 'some_function'2005-03-23 23:47:11,673 - spam_application - INFO -   done with auxiliary_module.some_function()

importloggingimportthreadingimporttimedefworker(arg):whilenotarg['stop']:logging.debug('Hi from myfunc')time.sleep(0.5)defmain():logging.basicConfig(level=logging.DEBUG,format='%(relativeCreated)6d%(threadName)s%(message)s')info={'stop':False}thread=threading.Thread(target=worker,args=(info,))thread.start()whileTrue:try:logging.debug('Hello from main')time.sleep(0.75)exceptKeyboardInterrupt:info['stop']=Truebreakthread.join()if__name__=='__main__':main()

   0 Thread-1 Hi from myfunc   3 MainThread Hello from main 505 Thread-1 Hi from myfunc 755 MainThread Hello from main1007 Thread-1 Hi from myfunc1507 MainThread Hello from main1508 Thread-1 Hi from myfunc2010 Thread-1 Hi from myfunc2258 MainThread Hello from main2512 Thread-1 Hi from myfunc3009 MainThread Hello from main3013 Thread-1 Hi from myfunc3515 Thread-1 Hi from myfunc3761 MainThread Hello from main4017 Thread-1 Hi from myfunc4513 MainThread Hello from main4518 Thread-1 Hi from myfunc

importlogginglogger=logging.getLogger('simple_example')logger.setLevel(logging.DEBUG)# create file handler which logs even debug messagesfh=logging.FileHandler('spam.log')fh.setLevel(logging.DEBUG)# create console handler with a higher log levelch=logging.StreamHandler()ch.setLevel(logging.ERROR)# create formatter and add it to the handlersformatter=logging.Formatter('%(asctime)s -%(name)s -%(levelname)s -%(message)s')ch.setFormatter(formatter)fh.setFormatter(formatter)# add the handlers to loggerlogger.addHandler(ch)logger.addHandler(fh)# 'application' codelogger.debug('debug message')logger.info('info message')logger.warning('warn message')logger.error('error message')logger.critical('critical message')

importlogging# set up logging to file - see previous section for more detailslogging.basicConfig(level=logging.DEBUG,format='%(asctime)s%(name)-12s%(levelname)-8s%(message)s',datefmt='%m-%d %H:%M',filename='/temp/myapp.log',filemode='w')# define a Handler which writes INFO messages or higher to the sys.stderrconsole=logging.StreamHandler()console.setLevel(logging.INFO)# set a format which is simpler for console useformatter=logging.Formatter('%(name)-12s:%(levelname)-8s%(message)s')# tell the handler to use this formatconsole.setFormatter(formatter)# add the handler to the root loggerlogging.getLogger('').addHandler(console)# Now, we can log to the root logger, or any other logger. First the root...logging.info('Jackdaws love my big sphinx of quartz.')# Now, define a couple of other loggers which might represent areas in your# application:logger1=logging.getLogger('myapp.area1')logger2=logging.getLogger('myapp.area2')logger1.debug('Quick zephyrs blow, vexing daft Jim.')logger1.info('How quickly daft jumping zebras vex.')logger2.warning('Jail zesty vixen who grabbed pay from quack.')logger2.error('The five boxing wizards jump quickly.')

root        : INFO     Jackdaws love my big sphinx of quartz.myapp.area1 : INFO     How quickly daft jumping zebras vex.myapp.area2 : WARNING  Jail zesty vixen who grabbed pay from quack.myapp.area2 : ERROR    The five boxing wizards jump quickly.

10-22 22:19 root         INFO     Jackdaws love my big sphinx of quartz.10-22 22:19 myapp.area1  DEBUG    Quick zephyrs blow, vexing daft Jim.10-22 22:19 myapp.area1  INFO     How quickly daft jumping zebras vex.10-22 22:19 myapp.area2  WARNING  Jail zesty vixen who grabbed pay from quack.10-22 22:19 myapp.area2  ERROR    The five boxing wizards jump quickly.

importloggingimportlogging.configimporttimeimportos# read initial config filelogging.config.fileConfig('logging.conf')# create and start listener on port 9999t=logging.config.listen(9999)t.start()logger=logging.getLogger('simpleExample')try:# loop through logging calls to see the difference# new configurations make, until Ctrl+C is pressedwhileTrue:logger.debug('debug message')logger.info('info message')logger.warning('warn message')logger.error('error message')logger.critical('critical message')time.sleep(5)exceptKeyboardInterrupt:# cleanuplogging.config.stopListening()t.join()

#!/usr/bin/env pythonimportsocket,sys,structwithopen(sys.argv[1],'rb')asf:data_to_send=f.read()HOST='localhost'PORT=9999s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)print('connecting...')s.connect((HOST,PORT))print('sending config...')s.send(struct.pack('>L',len(data_to_send)))s.send(data_to_send)s.close()print('complete')

que=queue.Queue(-1)# no limit on sizequeue_handler=QueueHandler(que)handler=logging.StreamHandler()listener=QueueListener(que,handler)root=logging.getLogger()root.addHandler(queue_handler)formatter=logging.Formatter('%(threadName)s:%(message)s')handler.setFormatter(formatter)listener.start()# The log output will display the thread which generated# the event (the main thread) rather than the internal# thread which monitors the internal queue. This is what# you want to happen.root.warning('Look out!')listener.stop()

MainThread: Look out!

importlogging,logging.handlersrootLogger=logging.getLogger('')rootLogger.setLevel(logging.DEBUG)socketHandler=logging.handlers.SocketHandler('localhost',logging.handlers.DEFAULT_TCP_LOGGING_PORT)# don't bother with a formatter, since a socket handler sends the event as# an unformatted picklerootLogger.addHandler(socketHandler)# Now, we can log to the root logger, or any other logger. First the root...logging.info('Jackdaws love my big sphinx of quartz.')# Now, define a couple of other loggers which might represent areas in your# application:logger1=logging.getLogger('myapp.area1')logger2=logging.getLogger('myapp.area2')logger1.debug('Quick zephyrs blow, vexing daft Jim.')logger1.info('How quickly daft jumping zebras vex.')logger2.warning('Jail zesty vixen who grabbed pay from quack.')logger2.error('The five boxing wizards jump quickly.')

importpickleimportloggingimportlogging.handlersimportsocketserverimportstructclassLogRecordStreamHandler(socketserver.StreamRequestHandler):"""Handler for a streaming logging request.    This basically logs the record using whatever logging policy is    configured locally.    """defhandle(self):"""        Handle multiple requests - each expected to be a 4-byte length,        followed by the LogRecord in pickle format. Logs the record        according to whatever policy is configured locally.        """whileTrue:chunk=self.connection.recv(4)iflen(chunk)<4:breakslen=struct.unpack('>L',chunk)[0]chunk=self.connection.recv(slen)whilelen(chunk)<slen:chunk=chunk+self.connection.recv(slen-len(chunk))obj=self.unPickle(chunk)record=logging.makeLogRecord(obj)self.handleLogRecord(record)defunPickle(self,data):returnpickle.loads(data)defhandleLogRecord(self,record):# if a name is specified, we use the named logger rather than the one# implied by the record.ifself.server.lognameisnotNone:name=self.server.lognameelse:name=record.namelogger=logging.getLogger(name)# N.B. EVERY record gets logged. This is because Logger.handle# is normally called AFTER logger-level filtering. If you want# to do filtering, do it at the client end to save wasting# cycles and network bandwidth!logger.handle(record)classLogRecordSocketReceiver(socketserver.ThreadingTCPServer):"""    Simple TCP socket-based logging receiver suitable for testing.    """allow_reuse_address=Truedef__init__(self,host='localhost',port=logging.handlers.DEFAULT_TCP_LOGGING_PORT,handler=LogRecordStreamHandler):socketserver.ThreadingTCPServer.__init__(self,(host,port),handler)self.abort=0self.timeout=1self.logname=Nonedefserve_until_stopped(self):importselectabort=0whilenotabort:rd,wr,ex=select.select([self.socket.fileno()],[],[],self.timeout)ifrd:self.handle_request()abort=self.abortdefmain():logging.basicConfig(format='%(relativeCreated)5d%(name)-15s%(levelname)-8s%(message)s')tcpserver=LogRecordSocketReceiver()print('About to start TCP server...')tcpserver.serve_until_stopped()if__name__=='__main__':main()

About to start TCP server...   59 root            INFO     Jackdaws love my big sphinx of quartz.   59 myapp.area1     DEBUG    Quick zephyrs blow, vexing daft Jim.   69 myapp.area1     INFO     How quickly daft jumping zebras vex.   69 myapp.area2     WARNING  Jail zesty vixen who grabbed pay from quack.   69 myapp.area2     ERROR    The five boxing wizards jump quickly.

# You'll need these imports in your own codeimportloggingimportlogging.handlersimportmultiprocessing# Next two import lines for this demo onlyfromrandomimportchoice,randomimporttime## Because you'll want to define the logging configurations for listener and workers, the# listener and worker process functions take a configurer parameter which is a callable# for configuring logging for that process. These functions are also passed the queue,# which they use for communication.## In practice, you can configure the listener however you want, but note that in this# simple example, the listener does not apply level or filter logic to received records.# In practice, you would probably want to do this logic in the worker processes, to avoid# sending events which would be filtered out between processes.## The size of the rotated files is made small so you can see the results easily.deflistener_configurer():root=logging.getLogger()h=logging.handlers.RotatingFileHandler('mptest.log','a',300,10)f=logging.Formatter('%(asctime)s%(processName)-10s%(name)s%(levelname)-8s%(message)s')h.setFormatter(f)root.addHandler(h)# This is the listener process top-level loop: wait for logging events# (LogRecords)on the queue and handle them, quit when you get a None for a# LogRecord.deflistener_process(queue,configurer):configurer()whileTrue:try:record=queue.get()ifrecordisNone:# We send this as a sentinel to tell the listener to quit.breaklogger=logging.getLogger(record.name)logger.handle(record)# No level or filter logic applied - just do it!exceptException:importsys,tracebackprint('Whoops! Problem:',file=sys.stderr)traceback.print_exc(file=sys.stderr)# Arrays used for random selections in this demoLEVELS=[logging.DEBUG,logging.INFO,logging.WARNING,logging.ERROR,logging.CRITICAL]LOGGERS=['a.b.c','d.e.f']MESSAGES=['Random message #1','Random message #2','Random message #3',]# The worker configuration is done at the start of the worker process run.# Note that on Windows you can't rely on fork semantics, so each process# will run the logging configuration code when it starts.defworker_configurer(queue):h=logging.handlers.QueueHandler(queue)# Just the one handler neededroot=logging.getLogger()root.addHandler(h)# send all messages, for demo; no other level or filter logic applied.root.setLevel(logging.DEBUG)# This is the worker process top-level loop, which just logs ten events with# random intervening delays before terminating.# The print messages are just so you know it's doing something!defworker_process(queue,configurer):configurer(queue)name=multiprocessing.current_process().nameprint('Worker started:%s'%name)foriinrange(10):time.sleep(random())logger=logging.getLogger(choice(LOGGERS))level=choice(LEVELS)message=choice(MESSAGES)logger.log(level,message)print('Worker finished:%s'%name)# Here's where the demo gets orchestrated. Create the queue, create and start# the listener, create ten workers and start them, wait for them to finish,# then send a None to the queue to tell the listener to finish.defmain():queue=multiprocessing.Queue(-1)listener=multiprocessing.Process(target=listener_process,args=(queue,listener_configurer))listener.start()workers=[]foriinrange(10):worker=multiprocessing.Process(target=worker_process,args=(queue,worker_configurer))workers.append(worker)worker.start()forwinworkers:w.join()queue.put_nowait(None)listener.join()if__name__=='__main__':main()

importloggingimportlogging.configimportlogging.handlersfrommultiprocessingimportProcess,Queueimportrandomimportthreadingimporttimedeflogger_thread(q):whileTrue:record=q.get()ifrecordisNone:breaklogger=logging.getLogger(record.name)logger.handle(record)defworker_process(q):qh=logging.handlers.QueueHandler(q)root=logging.getLogger()root.setLevel(logging.DEBUG)root.addHandler(qh)levels=[logging.DEBUG,logging.INFO,logging.WARNING,logging.ERROR,logging.CRITICAL]loggers=['foo','foo.bar','foo.bar.baz','spam','spam.ham','spam.ham.eggs']foriinrange(100):lvl=random.choice(levels)logger=logging.getLogger(random.choice(loggers))logger.log(lvl,'Message no.%d',i)if__name__=='__main__':q=Queue()d={'version':1,'formatters':{'detailed':{'class':'logging.Formatter','format':'%(asctime)s%(name)-15s%(levelname)-8s%(processName)-10s%(message)s'}},'handlers':{'console':{'class':'logging.StreamHandler','level':'INFO',},'file':{'class':'logging.FileHandler','filename':'mplog.log','mode':'w','formatter':'detailed',},'foofile':{'class':'logging.FileHandler','filename':'mplog-foo.log','mode':'w','formatter':'detailed',},'errors':{'class':'logging.FileHandler','filename':'mplog-errors.log','mode':'w','level':'ERROR','formatter':'detailed',},},'loggers':{'foo':{'handlers':['foofile']}},'root':{'level':'DEBUG','handlers':['console','file','errors']},}workers=[]foriinrange(5):wp=Process(target=worker_process,name='worker%d'%(i+1),args=(q,))workers.append(wp)wp.start()logging.config.dictConfig(d)lp=threading.Thread(target=logger_thread,args=(q,))lp.start()# At this point, the main process could do some useful work of its own# Once it's done that, it can wait for the workers to terminate...forwpinworkers:wp.join()# And now tell the logging thread to finish up, tooq.put(None)lp.join()

importglobimportloggingimportlogging.handlersLOG_FILENAME='logging_rotatingfile_example.out'# Set up a specific logger with our desired output levelmy_logger=logging.getLogger('MyLogger')my_logger.setLevel(logging.DEBUG)# Add the log message handler to the loggerhandler=logging.handlers.RotatingFileHandler(LOG_FILENAME,maxBytes=20,backupCount=5)my_logger.addHandler(handler)# Log some messagesforiinrange(20):my_logger.debug('i =%d'%i)# See what files are createdlogfiles=glob.glob('%s*'%LOG_FILENAME)forfilenameinlogfiles:print(filename)

logging_rotatingfile_example.outlogging_rotatingfile_example.out.1logging_rotatingfile_example.out.2logging_rotatingfile_example.out.3logging_rotatingfile_example.out.4logging_rotatingfile_example.out.5

>>>importlogging>>>root=logging.getLogger()>>>root.setLevel(logging.DEBUG)>>>handler=logging.StreamHandler()>>>bf=logging.Formatter('{asctime}{name}{levelname:8s}{message}',...style='{')>>>handler.setFormatter(bf)>>>root.addHandler(handler)>>>logger=logging.getLogger('foo.bar')>>>logger.debug('This is a DEBUG message')2010-10-28 15:11:55,341 foo.bar DEBUG    This is a DEBUG message>>>logger.critical('This is a CRITICAL message')2010-10-28 15:12:11,526 foo.bar CRITICAL This is a CRITICAL message>>>df=logging.Formatter('$asctime $name ${levelname} $message',...style='$')>>>handler.setFormatter(df)>>>logger.debug('This is a DEBUG message')2010-10-28 15:13:06,924 foo.bar DEBUG This is a DEBUG message>>>logger.critical('This is a CRITICAL message')2010-10-28 15:13:11,494 foo.bar CRITICAL This is a CRITICAL message>>>

>>>logger.error('This is an%s%s%s','other,','ERROR,','message')2010-10-28 15:19:29,833 foo.bar ERROR This is another, ERROR, message>>>

classBraceMessage:def__init__(self,fmt,/,*args,**kwargs):self.fmt=fmtself.args=argsself.kwargs=kwargsdef__str__(self):returnself.fmt.format(*self.args,**self.kwargs)classDollarMessage:def__init__(self,fmt,/,**kwargs):self.fmt=fmtself.kwargs=kwargsdef__str__(self):fromstringimportTemplatereturnTemplate(self.fmt).substitute(**self.kwargs)

>>>fromwhereverimportBraceMessageas__>>>print(__('Message with{0}{name}',2,name='placeholders'))Message with 2 placeholders>>>classPoint:pass...>>>p=Point()>>>p.x=0.5>>>p.y=0.5>>>print(__('Message with coordinates: ({point.x:.2f},{point.y:.2f})',...point=p))Message with coordinates: (0.50, 0.50)>>>fromwhereverimportDollarMessageas__>>>print(__('Message with $num $what',num=2,what='placeholders'))Message with 2 placeholders>>>

importloggingclassMessage:def__init__(self,fmt,args):self.fmt=fmtself.args=argsdef__str__(self):returnself.fmt.format(*self.args)classStyleAdapter(logging.LoggerAdapter):def__init__(self,logger,extra=None):super().__init__(logger,extraor{})deflog(self,level,msg,/,*args,**kwargs):ifself.isEnabledFor(level):msg,kwargs=self.process(msg,kwargs)self.logger._log(level,Message(msg,args),(),**kwargs)logger=StyleAdapter(logging.getLogger(__name__))defmain():logger.debug('Hello,{}','world!')if__name__=='__main__':logging.basicConfig(level=logging.DEBUG)main()

logger=logging.getLogger(__name__)

old_factory=logging.getLogRecordFactory()defrecord_factory(*args,**kwargs):record=old_factory(*args,**kwargs)record.custom_attribute=0xdecafbadreturnrecordlogging.setLogRecordFactory(record_factory)

importzmq# using pyzmq, the Python binding for ZeroMQimportjson# for serializing records portablyctx=zmq.Context()sock=zmq.Socket(ctx,zmq.PUB)# or zmq.PUSH, or other suitable valuesock.bind('tcp://*:5556')# or whereverclassZeroMQSocketHandler(QueueHandler):defenqueue(self,record):self.queue.send_json(record.__dict__)handler=ZeroMQSocketHandler(sock)

classZeroMQSocketHandler(QueueHandler):def__init__(self,uri,socktype=zmq.PUB,ctx=None):self.ctx=ctxorzmq.Context()socket=zmq.Socket(self.ctx,socktype)socket.bind(uri)super().__init__(socket)defenqueue(self,record):self.queue.send_json(record.__dict__)defclose(self):self.queue.close()

classZeroMQSocketListener(QueueListener):def__init__(self,uri,/,*handlers,**kwargs):self.ctx=kwargs.get('ctx')orzmq.Context()socket=zmq.Socket(self.ctx,zmq.SUB)socket.setsockopt_string(zmq.SUBSCRIBE,'')# subscribe to everythingsocket.connect(uri)super().__init__(socket,*handlers,**kwargs)defdequeue(self):msg=self.queue.recv_json()returnlogging.makeLogRecord(msg)

LOGGING={'version':1,'disable_existing_loggers':True,'formatters':{'verbose':{'format':'%(levelname)s%(asctime)s%(module)s%(process)d%(thread)d%(message)s'},'simple':{'format':'%(levelname)s%(message)s'},},'filters':{'special':{'()':'project.logging.SpecialFilter','foo':'bar',}},'handlers':{'null':{'level':'DEBUG','class':'django.utils.log.NullHandler',},'console':{'level':'DEBUG','class':'logging.StreamHandler','formatter':'simple'},'mail_admins':{'level':'ERROR','class':'django.utils.log.AdminEmailHandler','filters':['special']}},'loggers':{'django':{'handlers':['null'],'propagate':True,'level':'INFO',},'django.request':{'handlers':['mail_admins'],'level':'ERROR','propagate':False,},'myproject.custom':{'handlers':['console','mail_admins'],'level':'INFO','filters':['special']}}}

defnamer(name):returnname+".gz"defrotator(source,dest):withopen(source,"rb")assf:data=sf.read()compressed=zlib.compress(data,9)withopen(dest,"wb")asdf:df.write(compressed)os.remove(source)rh=logging.handlers.RotatingFileHandler(...)rh.rotator=rotatorrh.namer=namer

importloggingimportlogging.configimportlogging.handlersfrommultiprocessingimportProcess,Queue,Event,current_processimportosimportrandomimporttimeclassMyHandler:"""    A simple handler for logging events. It runs in the listener process and    dispatches events to loggers based on the name in the received record,    which then get dispatched, by the logging system, to the handlers    configured for those loggers.    """defhandle(self,record):ifrecord.name=="root":logger=logging.getLogger()else:logger=logging.getLogger(record.name)iflogger.isEnabledFor(record.levelno):# The process name is transformed just to show that it's the listener# doing the logging to files and consolerecord.processName='%s (for%s)'%(current_process().name,record.processName)logger.handle(record)deflistener_process(q,stop_event,config):"""    This could be done in the main process, but is just done in a separate    process for illustrative purposes.    This initialises logging according to the specified configuration,    starts the listener and waits for the main process to signal completion    via the event. The listener is then stopped, and the process exits.    """logging.config.dictConfig(config)listener=logging.handlers.QueueListener(q,MyHandler())listener.start()ifos.name=='posix':# On POSIX, the setup logger will have been configured in the# parent process, but should have been disabled following the# dictConfig call.# On Windows, since fork isn't used, the setup logger won't# exist in the child, so it would be created and the message# would appear - hence the "if posix" clause.logger=logging.getLogger('setup')logger.critical('Should not appear, because of disabled logger ...')stop_event.wait()listener.stop()defworker_process(config):"""    A number of these are spawned for the purpose of illustration. In    practice, they could be a heterogeneous bunch of processes rather than    ones which are identical to each other.    This initialises logging according to the specified configuration,    and logs a hundred messages with random levels to randomly selected    loggers.    A small sleep is added to allow other processes a chance to run. This    is not strictly needed, but it mixes the output from the different    processes a bit more than if it's left out.    """logging.config.dictConfig(config)levels=[logging.DEBUG,logging.INFO,logging.WARNING,logging.ERROR,logging.CRITICAL]loggers=['foo','foo.bar','foo.bar.baz','spam','spam.ham','spam.ham.eggs']ifos.name=='posix':# On POSIX, the setup logger will have been configured in the# parent process, but should have been disabled following the# dictConfig call.# On Windows, since fork isn't used, the setup logger won't# exist in the child, so it would be created and the message# would appear - hence the "if posix" clause.logger=logging.getLogger('setup')logger.critical('Should not appear, because of disabled logger ...')foriinrange(100):lvl=random.choice(levels)logger=logging.getLogger(random.choice(loggers))logger.log(lvl,'Message no.%d',i)time.sleep(0.01)defmain():q=Queue()# The main process gets a simple configuration which prints to the console.config_initial={'version':1,'handlers':{'console':{'class':'logging.StreamHandler','level':'INFO'}},'root':{'handlers':['console'],'level':'DEBUG'}}# The worker process configuration is just a QueueHandler attached to the# root logger, which allows all messages to be sent to the queue.# We disable existing loggers to disable the "setup" logger used in the# parent process. This is needed on POSIX because the logger will# be there in the child following a fork().config_worker={'version':1,'disable_existing_loggers':True,'handlers':{'queue':{'class':'logging.handlers.QueueHandler','queue':q}},'root':{'handlers':['queue'],'level':'DEBUG'}}# The listener process configuration shows that the full flexibility of# logging configuration is available to dispatch events to handlers however# you want.# We disable existing loggers to disable the "setup" logger used in the# parent process. This is needed on POSIX because the logger will# be there in the child following a fork().config_listener={'version':1,'disable_existing_loggers':True,'formatters':{'detailed':{'class':'logging.Formatter','format':'%(asctime)s%(name)-15s%(levelname)-8s%(processName)-10s%(message)s'},'simple':{'class':'logging.Formatter','format':'%(name)-15s%(levelname)-8s%(processName)-10s%(message)s'}},'handlers':{'console':{'class':'logging.StreamHandler','formatter':'simple','level':'INFO'},'file':{'class':'logging.FileHandler','filename':'mplog.log','mode':'w','formatter':'detailed'},'foofile':{'class':'logging.FileHandler','filename':'mplog-foo.log','mode':'w','formatter':'detailed'},'errors':{'class':'logging.FileHandler','filename':'mplog-errors.log','mode':'w','formatter':'detailed','level':'ERROR'}},'loggers':{'foo':{'handlers':['foofile']}},'root':{'handlers':['console','file','errors'],'level':'DEBUG'}}# Log some initial events, just to show that logging in the parent works# normally.logging.config.dictConfig(config_initial)logger=logging.getLogger('setup')logger.info('About to create workers ...')workers=[]foriinrange(5):wp=Process(target=worker_process,name='worker%d'%(i+1),args=(config_worker,))workers.append(wp)wp.start()logger.info('Started worker:%s',wp.name)logger.info('About to create listener ...')stop_event=Event()lp=Process(target=listener_process,name='listener',args=(q,stop_event,config_listener))lp.start()logger.info('Started listener')# We now hang around for the workers to finish their work.forwpinworkers:wp.join()# Workers all done, listening can now stop.# Logging in the parent still works normally.logger.info('Telling listener to stop ...')stop_event.set()lp.join()logger.info('All done.')if__name__=='__main__':main()

importjsonimportloggingclassStructuredMessage:def__init__(self,message,/,**kwargs):self.message=messageself.kwargs=kwargsdef__str__(self):return'%s >>>%s'%(self.message,json.dumps(self.kwargs))_=StructuredMessage# optional, to improve readabilitylogging.basicConfig(level=logging.INFO,format='%(message)s')logging.info(_('message 1',foo='bar',bar='baz',num=123,fnum=123.456))

message 1 >>> {"fnum": 123.456, "num": 123, "bar": "baz", "foo": "bar"}

from__future__importunicode_literalsimportjsonimportlogging# This next bit is to ensure the script runs unchanged on 2.x and 3.xtry:unicodeexceptNameError:unicode=strclassEncoder(json.JSONEncoder):defdefault(self,o):ifisinstance(o,set):returntuple(o)elifisinstance(o,unicode):returno.encode('unicode_escape').decode('ascii')returnsuper().default(o)classStructuredMessage:def__init__(self,message,/,**kwargs):self.message=messageself.kwargs=kwargsdef__str__(self):s=Encoder().encode(self.kwargs)return'%s >>>%s'%(self.message,s)_=StructuredMessage# optional, to improve readabilitydefmain():logging.basicConfig(level=logging.INFO,format='%(message)s')logging.info(_('message 1',set_value={1,2,3},snowman='\u2603'))if__name__=='__main__':main()

message 1 >>> {"snowman": "\u2603", "set_value": [1, 2, 3]}

defowned_file_handler(filename,mode='a',encoding=None,owner=None):ifowner:ifnotos.path.exists(filename):open(filename,'a').close()shutil.chown(filename,*owner)returnlogging.FileHandler(filename,mode,encoding)

LOGGING={'version':1,'disable_existing_loggers':False,'formatters':{'default':{'format':'%(asctime)s%(levelname)s%(name)s%(message)s'},},'handlers':{'file':{# The values below are popped from this dictionary and# used to create the handler, set the handler's level and# its formatter.'()':owned_file_handler,'level':'DEBUG','formatter':'default',# The values below are passed to the handler creator callable# as keyword arguments.'owner':['pulse','pulse'],'filename':'chowntest.log','mode':'w','encoding':'utf-8',},},'root':{'handlers':['file'],'level':'DEBUG',},}

importlogging,logging.config,os,shutildefowned_file_handler(filename,mode='a',encoding=None,owner=None):ifowner:ifnotos.path.exists(filename):open(filename,'a').close()shutil.chown(filename,*owner)returnlogging.FileHandler(filename,mode,encoding)LOGGING={'version':1,'disable_existing_loggers':False,'formatters':{'default':{'format':'%(asctime)s%(levelname)s%(name)s%(message)s'},},'handlers':{'file':{# The values below are popped from this dictionary and# used to create the handler, set the handler's level and# its formatter.'()':owned_file_handler,'level':'DEBUG','formatter':'default',# The values below are passed to the handler creator callable# as keyword arguments.'owner':['pulse','pulse'],'filename':'chowntest.log','mode':'w','encoding':'utf-8',},},'root':{'handlers':['file'],'level':'DEBUG',},}logging.config.dictConfig(LOGGING)logger=logging.getLogger('mylogger')logger.debug('A debug message')

$sudopython3.3chowntest.py$catchowntest.log2013-11-05 09:34:51,128 DEBUG mylogger A debug message$ls-lchowntest.log-rw-r--r-- 1 pulse pulse 55 2013-11-05 09:34 chowntest.log

'()':owned_file_handler,

'()':'ext://project.util.owned_file_handler',

importloggingimportlogging.configimportsysclassMyFilter(logging.Filter):def__init__(self,param=None):self.param=paramdeffilter(self,record):ifself.paramisNone:allow=Trueelse:allow=self.paramnotinrecord.msgifallow:record.msg='changed: '+record.msgreturnallowLOGGING={'version':1,'filters':{'myfilter':{'()':MyFilter,'param':'noshow',}},'handlers':{'console':{'class':'logging.StreamHandler','filters':['myfilter']}},'root':{'level':'DEBUG','handlers':['console']},}if__name__=='__main__':logging.config.dictConfig(LOGGING)logging.debug('hello')logging.debug('hello - noshow')

changed: hello

importloggingclassOneLineExceptionFormatter(logging.Formatter):defformatException(self,exc_info):"""        Format an exception so that it prints on a single line.        """result=super().formatException(exc_info)returnrepr(result)# or format into one line however you want todefformat(self,record):s=super().format(record)ifrecord.exc_text:s=s.replace('\n','')+'|'returnsdefconfigure_logging():fh=logging.FileHandler('output.txt','w')f=OneLineExceptionFormatter('%(asctime)s|%(levelname)s|%(message)s|','%d/%m/%Y %H:%M:%S')fh.setFormatter(f)root=logging.getLogger()root.setLevel(logging.DEBUG)root.addHandler(fh)defmain():configure_logging()logging.info('Sample message')try:x=1/0exceptZeroDivisionErrorase:logging.exception('ZeroDivisionError:%s',e)if__name__=='__main__':main()

28/01/2015 07:21:23|INFO|Sample message|28/01/2015 07:21:23|ERROR|ZeroDivisionError: integer division or modulo by zero|'Traceback (most recent call last):\n  File "logtest7.py", line 30, in main\n    x = 1 / 0\nZeroDivisionError: integer division or modulo by zero'|

importloggingimportsubprocessimportsysclassTTSHandler(logging.Handler):defemit(self,record):msg=self.format(record)# Speak slowly in a female English voicecmd=['espeak','-s150','-ven+f3',msg]p=subprocess.Popen(cmd,stdout=subprocess.PIPE,stderr=subprocess.STDOUT)# wait for the program to finishp.communicate()defconfigure_logging():h=TTSHandler()root=logging.getLogger()root.addHandler(h)# the default formatter just returns the messageroot.setLevel(logging.DEBUG)defmain():logging.info('Hello')logging.debug('Goodbye')if__name__=='__main__':configure_logging()sys.exit(main())

importloggingfromlogging.handlersimportMemoryHandlerimportsyslogger=logging.getLogger(__name__)logger.addHandler(logging.NullHandler())deflog_if_errors(logger,target_handler=None,flush_level=None,capacity=None):iftarget_handlerisNone:target_handler=logging.StreamHandler()ifflush_levelisNone:flush_level=logging.ERRORifcapacityisNone:capacity=100handler=MemoryHandler(capacity,flushLevel=flush_level,target=target_handler)defdecorator(fn):defwrapper(*args,**kwargs):logger.addHandler(handler)try:returnfn(*args,**kwargs)exceptException:logger.exception('call failed')raisefinally:super(MemoryHandler,handler).flush()logger.removeHandler(handler)returnwrapperreturndecoratordefwrite_line(s):sys.stderr.write('%s\n'%s)deffoo(fail=False):write_line('about to log at DEBUG ...')logger.debug('Actually logged at DEBUG')write_line('about to log at INFO ...')logger.info('Actually logged at INFO')write_line('about to log at WARNING ...')logger.warning('Actually logged at WARNING')iffail:write_line('about to log at ERROR ...')logger.error('Actually logged at ERROR')write_line('about to log at CRITICAL ...')logger.critical('Actually logged at CRITICAL')returnfaildecorated_foo=log_if_errors(logger)(foo)if__name__=='__main__':logger.setLevel(logging.DEBUG)write_line('Calling undecorated foo with False')assertnotfoo(False)write_line('Calling undecorated foo with True')assertfoo(True)write_line('Calling decorated foo with False')assertnotdecorated_foo(False)write_line('Calling decorated foo with True')assertdecorated_foo(True)

Calling undecorated foo with Falseabout to log at DEBUG ...about to log at INFO ...about to log at WARNING ...Calling undecorated foo with Trueabout to log at DEBUG ...about to log at INFO ...about to log at WARNING ...about to log at ERROR ...about to log at CRITICAL ...Calling decorated foo with Falseabout to log at DEBUG ...about to log at INFO ...about to log at WARNING ...Calling decorated foo with Trueabout to log at DEBUG ...about to log at INFO ...about to log at WARNING ...about to log at ERROR ...Actually logged at DEBUGActually logged at INFOActually logged at WARNINGActually logged at ERRORabout to log at CRITICAL ...Actually logged at CRITICAL

@log_if_errors(logger)deffoo(fail=False):...

importloggingimporttimeclassUTCFormatter(logging.Formatter):converter=time.gmtime

importloggingimportlogging.configimporttimeclassUTCFormatter(logging.Formatter):converter=time.gmtimeLOGGING={'version':1,'disable_existing_loggers':False,'formatters':{'utc':{'()':UTCFormatter,'format':'%(asctime)s%(message)s',},'local':{'format':'%(asctime)s%(message)s',}},'handlers':{'console1':{'class':'logging.StreamHandler','formatter':'utc',},'console2':{'class':'logging.StreamHandler','formatter':'local',},},'root':{'handlers':['console1','console2'],}}if__name__=='__main__':logging.config.dictConfig(LOGGING)logging.warning('The local time is%s',time.asctime())

2015-10-17 12:53:29,501 The local time is Sat Oct 17 13:53:29 20152015-10-17 13:53:29,501 The local time is Sat Oct 17 13:53:29 2015

importloggingimportsysclassLoggingContext:def__init__(self,logger,level=None,handler=None,close=True):self.logger=loggerself.level=levelself.handler=handlerself.close=closedef__enter__(self):ifself.levelisnotNone:self.old_level=self.logger.levelself.logger.setLevel(self.level)ifself.handler:self.logger.addHandler(self.handler)def__exit__(self,et,ev,tb):ifself.levelisnotNone:self.logger.setLevel(self.old_level)ifself.handler:self.logger.removeHandler(self.handler)ifself.handlerandself.close:self.handler.close()# implicit return of None => don't swallow exceptions

if__name__=='__main__':logger=logging.getLogger('foo')logger.addHandler(logging.StreamHandler())logger.setLevel(logging.INFO)logger.info('1. This should appear just once on stderr.')logger.debug('2. This should not appear.')withLoggingContext(logger,level=logging.DEBUG):logger.debug('3. This should appear once on stderr.')logger.debug('4. This should not appear.')h=logging.StreamHandler(sys.stdout)withLoggingContext(logger,level=logging.DEBUG,handler=h,close=True):logger.debug('5. This should appear twice - once on stderr and once on stdout.')logger.info('6. This should appear just once on stderr.')logger.debug('7. This should not appear.')

$pythonlogctx.py1. This should appear just once on stderr.3. This should appear once on stderr.5. This should appear twice - once on stderr and once on stdout.5. This should appear twice - once on stderr and once on stdout.6. This should appear just once on stderr.

$pythonlogctx.py2>/dev/null5. This should appear twice - once on stderr and once on stdout.

$pythonlogctx.py>/dev/null1. This should appear just once on stderr.3. This should appear once on stderr.5. This should appear twice - once on stderr and once on stdout.6. This should appear just once on stderr.

importargparseimportimportlibimportloggingimportosimportsysdefmain(args=None):scriptname=os.path.basename(__file__)parser=argparse.ArgumentParser(scriptname)levels=('DEBUG','INFO','WARNING','ERROR','CRITICAL')parser.add_argument('--log-level',default='INFO',choices=levels)subparsers=parser.add_subparsers(dest='command',help='Available commands:')start_cmd=subparsers.add_parser('start',help='Start a service')start_cmd.add_argument('name',metavar='NAME',help='Name of service to start')stop_cmd=subparsers.add_parser('stop',help='Stop one or more services')stop_cmd.add_argument('names',metavar='NAME',nargs='+',help='Name of service to stop')restart_cmd=subparsers.add_parser('restart',help='Restart one or more services')restart_cmd.add_argument('names',metavar='NAME',nargs='+',help='Name of service to restart')options=parser.parse_args()# the code to dispatch commands could all be in this file. For the purposes# of illustration only, we implement each command in a separate module.try:mod=importlib.import_module(options.command)cmd=getattr(mod,'command')except(ImportError,AttributeError):print('Unable to find the code for command\'%s\''%options.command)return1# Could get fancy here and load configuration from file or dictionarylogging.basicConfig(level=options.log_level,format='%(levelname)s%(name)s%(message)s')cmd(options)if__name__=='__main__':sys.exit(main())

# start.pyimportlogginglogger=logging.getLogger(__name__)defcommand(options):logger.debug('About to start%s',options.name)# actually do the command processing here ...logger.info('Started the\'%s\' service.',options.name)

# stop.pyimportlogginglogger=logging.getLogger(__name__)defcommand(options):n=len(options.names)ifn==1:plural=''services='\'%s\''%options.names[0]else:plural='s'services=', '.join('\'%s\''%namefornameinoptions.names)i=services.rfind(', ')services=services[:i]+' and '+services[i+2:]logger.debug('About to stop%s',services)# actually do the command processing here ...logger.info('Stopped the%s service%s.',services,plural)

# restart.pyimportlogginglogger=logging.getLogger(__name__)defcommand(options):n=len(options.names)ifn==1:plural=''services='\'%s\''%options.names[0]else:plural='s'services=', '.join('\'%s\''%namefornameinoptions.names)i=services.rfind(', ')services=services[:i]+' and '+services[i+2:]logger.debug('About to restart%s',services)# actually do the command processing here ...logger.info('Restarted the%s service%s.',services,plural)