python更加灵活的Logger日志详解

#!/usr/bin/env python3
# coding=utf-8
 
import logging
from logging import handlers
import time
 
log = logging.getLogger("mylog")
log.setLevel(logging.DEBUG)
h = logging.handlers.SocketHandler(host="127.0.0.1", port=8899)
f = logging.Formatter("[%(name)s][%(asctime)s]%(message)s")
h.setFormatter(f)
log.addHandler(h)
for x in range(1, 6):
    log.info("test log %d" % x)
    time.sleep(0.5)

#!/usr/bin/env Python3
# coding=utf-8
 
import socket
import pickle
import struct
import logging
 
 
def unPickle(bs: bytes):
    data_len_bytes = bs[0:4]
    data_len = struct.unpack(">L", data_len_bytes)[0]
    pickled_data = bs[4: data_len + 4 + 1]
    return pickle.loads(pickled_data)
 
 
socket_server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket_server.bind(("127.0.0.1", 8899))
socket_server.listen(3)
while True:
    try:
        client, client_addr = socket_server.accept()
        while True:
            data = client.recv(1024 * 10)
            if data != b'':
                log_dict = unPickle(data)
                log_record = logging.makeLogRecord(log_dict)
                print("recv log:", log_record)
    except Exception as e:
        print("Exception:", e)
 

#!/usr/bin/env python3
# coding=utf-8
 
import logging
from logging import handlers
import time
 
log = logging.getLogger("mylog")
log.setLevel(logging.DEBUG)
h_get = handlers.HTTPHandler(host="127.0.0.1:8080", url="test_http_log", method="GET",
                             secure=False, credentials=None, context=None)
log.addHandler(h_get)
h_post = handlers.HTTPHandler(host="127.0.0.1:8080", url="test_http_log", method="POST",
                              secure=False, credentials=None, context=None)
log.addHandler(h_post)
 
h = logging.StreamHandler()
f = logging.Formatter("[%(levelname)s][%(asctime)s]%(message)s")
h.setFormatter(f)
log.addHandler(h)
for x in range(1, 3):
    log.info("test HTTP log %d" % x)
    time.sleep(0.5)

#!/usr/bin/env python3
# coding=utf-8
 
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
from urllib import parse
import logging
import time
 
 
class MyHttpServer(BaseHTTPRequestHandler):
    def do_GET(self):
        url = parse.urlparse(self.path)
        q = parse.parse_qs(url.query)
        log = logging.makeLogRecord(q)
        log_created = time.localtime(float(log.created[0]))
        format_time = time.strftime("%Y/%m/%d %H:%M:%S", log_created)
        print("-GET:", log.name, log.levelname, log.msg, format_time+","+str(int(float(log.msecs[0])//1)))
        self.send_response(200)
        self.end_headers()
        self.wfile.flush()
 
    def do_POST(self):
        length = int(self.headers["content-length"])
        data = self.rfile.read(length)
        data = data.decode(encoding="utf-8")
        data_dict = parse.parse_qs(data)
        data = logging.makeLogRecord(data_dict)
        print("=POST:", data)
        self.send_response(200)
        self.end_headers()
        self.wfile.flush()
 
 
httpserver = ThreadingHTTPServer(("127.0.0.1", 8080), MyHttpServer)
httpserver.serve_forever()

#!/usr/bin/env python3
# coding=utf-8
 
import logging
from logging import handlers
from logging import LogRecord
import queue
 
 
class MyLogHandler(logging.Handler):
    def handle(self, record: LogRecord) -> None:
        print(record)
 
 
log = logging.getLogger("mylog")
log.setLevel(logging.DEBUG)
q = queue.SimpleQueue()
h = handlers.QueueHandler(q)
f = logging.Formatter("[%(levelname)s]%(message)s")
h.setFormatter(f)
log.addHandler(h)
 
listener = handlers.QueueListener(q, MyLogHandler(), respect_handler_level=False)
 
listener.start()
print("listener started")
log.info("test queue log1 info")
log.error("test queue log2 error")
 
listener.stop()
print("listener stopted")
log.info("test queue log3 info")
log.error("test queue log4 error")
print("test end")

#!/usr/bin/env python3
# coding=utf-8
 
import logging
from logging import LogRecord
from logging import handlers
import sys
 
# log1 = logging.Logger("a", logging.DEBUG)
log1 = logging.getLogger("a")
_log1 = logging.getLogger("a")
print(log1 is _log1, id(log1), id(_log1))
log1.setLevel(logging.DEBUG)
log2 = log1.getChild("b")
log3 = log1.getChild("c")
print(log1)
print(log2)
print(log3)
 
h1 = logging.StreamHandler()
format1 = logging.Formatter("[H1] [%(levelname)-8s] %(message)s")
h1.setFormatter(format1)
log1.addHandler(h1)
 
h2 = logging.StreamHandler()
format2 = logging.Formatter(fmt="[H2][%(levelname)7s][%(asctime)s]%(message)s", datefmt="%c")
h2.setFormatter(format2)
 
 
class myFilter(logging.Filter):
    def filter(self, record: LogRecord) -> int:
        print("record:" + repr(record))
        if "HELLO" in record.msg:
            return True
        else:
            return False
 
 
h2.addFilter(myFilter())
log2.addHandler(h2)
# log2.addFilter(myFilter())
 
h3 = logging.StreamHandler()
format3 = logging.Formatter("[H3]{%(levelname)s}{%(name)s}%(message)s")
h3.setFormatter(format3)
h3.setStream(sys.stdout)
h3.setLevel(logging.INFO)
log3.addHandler(h3)
#
h3_file = logging.FileHandler(filename="test_Logging2.log", mode="w")
h3_file.setFormatter(logging.Formatter("[H3File][%(levelname)8s][%(asctime)s]%(message)s"))
log3.addHandler(h3_file)
#
h3_rotatingfile = handlers.RotatingFileHandler(filename="test_Logging2_rotating.log", mode="a",
                                               maxBytes=256, backupCount=3,
                                               encoding="utf-8", delay=False)
h3_rotatingfile.setFormatter(logging.Formatter("[h3_rotating][%(levelname)s] %(message)s"))
log3.addHandler(h3_rotatingfile)
#
h3_timedrotationgfile = handlers.TimedRotatingFileHandler(filename="test_Logging2_timedrotating.log",
                                                          when="S", interval=5,
                                                          backupCount=5, encoding="utf-8",
                                                          delay=False, utc=False,
                                                          # atTime=
                                                          )
h3_timedrotationgfile.setFormatter(logging.Formatter("[H3timed][%(levelname)s]%(message)s"))
log3.addHandler(h3_timedrotationgfile)
log3.propagate = False
 
log1.debug("test log1")
log1.warning("warn log1")
log2.debug("test log2 HELLO")
log2.info("info log2")
log2.warning("warn log2 HELLO")
log2.error("error log2")
log3.debug("test log3")
log3.warning("warn log3")