当前位置： 首页 > wzjs >正文

江苏省义务教育标准化建设网站天津抖音seo

wzjs 2025/8/17 1:38:23
江苏省义务教育标准化建设网站,天津抖音seo,wordpress+h5幻灯片,中卫网红美食打卡地在 Python 开发过程中，日志（Logging）是不可或缺的调试和监控工具。合理的日志管理不仅能帮助开发者快速定位问题，还能提供丰富的数据支持，让应用更具可观测性。本文将带你全面了解 Python logging 模块，涵盖…

在这里插入图片描述
在 Python 开发过程中，日志（Logging）是不可或缺的调试和监控工具。合理的日志管理不仅能帮助开发者快速定位问题，还能提供丰富的数据支持，让应用更具可观测性。本文将带你全面了解 Python logging 模块，涵盖日志输出到控制台、文件、Elasticsearch（ES），以及日志格式的介绍。

一、日志输出到控制台

最简单的日志管理方式就是将日志信息输出到控制台（标准输出）。这种方式在调试时非常方便，能够实时查看程序运行状态。默认情况下，Python logging 模块会以 WARNING 级别及以上的日志输出到控制台。

  def add_stream_handle(self):if LOG_TO_CONSOLE and LOG_ENABLED:handler = logging.StreamHandler(sys.stdout)handler.setLevel(level=CONSOLE_LEVEL)handler.setFormatter(logging.Formatter(LOG_FORMAT))self.logger.addHandler(handler)

二、日志输出到文件

在实际开发中，日志不仅仅是为了调试，更重要的是长期存储以便后续分析。因此，我们通常需要将日志写入文件，以便在程序运行后仍能追踪问题。通过 FileHandler，我们可以轻松地将日志写入指定的文件，并支持日志轮转（如按日期或文件大小分割）。

    def add_file_handle(self):if LOG_ENABLED and LOG_TO_FILE:logs_dir = os.path.join(LOG_DIR, "logs")os.makedirs(logs_dir, exist_ok=True)file = os.path.join(logs_dir, f'{datetime.now().strftime("%Y%m%d_%H%M%S")}.log')handler = logging.FileHandler(file, encoding="utf-8")handler.setLevel(level=FILE_LEVEL)handler.setFormatter(logging.Formatter(LOG_FORMAT))self.logger.addHandler(handler)

三、日志输出到 Elasticsearch（ES）

对于需要集中式日志管理的场景，可以将日志直接存储到 Elasticsearch 中，配合 Kibana 进行可视化分析。通过 logging 模块的自定义 Handler 或者 elastic-apm 等第三方库，我们可以将 Python 产生的日志数据直接推送到 ES，并结合全文检索和仪表盘进行日志分析。

class JsonSerializer:"""JSON serializer relying on the standard library json module."""mimetype: ClassVar[str] = "application/json"def default(self, data: Any) -> Any:if isinstance(data, date):return data.isoformat()elif isinstance(data, uuid.UUID):return str(data)elif isinstance(data, Decimal):return float(data)raise Exception(f"Unable to serialize to JSON: {data!r} (type: {type(data).__name__})", )def json_dumps(self, data: dict) -> str:return json.dumps(data, default=self.default, ensure_ascii=False, separators=(",", ":"))def json_loads(self, data: str) -> dict:return json.loads(data)def loads(self, data: str) -> dict:return self.json_loads(data)def dumps(self, data: dict) -> str:# The body is already encoded to bytes# so we forward the request body along.if isinstance(data, str):return datareturn self.json_dumps(data)def _attempt_serialize_numpy(data: Any) -> Tuple[bool, Any]:try:import numpy as npif isinstance(data,(np.int_,np.intc,np.int8,np.int16,np.int32,np.int64,np.uint8,np.uint16,np.uint32,np.uint64,),):return True, int(data)elif isinstance(data,(np.float16,np.float32,np.float64,),):return True, float(data)elif isinstance(data, np.bool_):return True, bool(data)elif isinstance(data, np.datetime64):return True, data.item().isoformat()elif isinstance(data, np.ndarray):return True, data.tolist()except ImportError:# Since we failed to import 'numpy' we don't want to try again.return False, Nonereturn False, Nonedef _attempt_serialize_pandas(data: Any) -> Tuple[bool, Any]:try:import pandas as pdif isinstance(data, (pd.Series, pd.Categorical)):return True, data.tolist()elif isinstance(data, pd.Timestamp) and data is not getattr(pd, "NaT", None):return True, data.isoformat()elif data is getattr(pd, "NA", None):return True, Noneexcept ImportError:# Since we failed to import 'pandas' we don't want to try again.return False, Nonereturn False, Nonedef _attempt_serialize_numpy_or_pandas(data: Any) -> Tuple[bool, Any]:serialized, value = _attempt_serialize_numpy(data)if serialized:return serialized, valueserialized, value = _attempt_serialize_pandas(data)if serialized:return serialized, valuereturn False, NoneTIME_TYPES = (date, datetime)
FLOAT_TYPES = (Decimal,)
INTEGER_TYPES = ()class EsJsonSerializer(JsonSerializer):mimetype: ClassVar[str] = "application/json"def default(self, data: Any) -> Any:if isinstance(data, TIME_TYPES):# Little hack to avoid importing pandas but to not# return 'NaT' string for pd.NaT as that's not a valid# Elasticsearch date.formatted_data = data.isoformat()if formatted_data != "NaT":return formatted_dataif isinstance(data, uuid.UUID):return str(data)elif isinstance(data, FLOAT_TYPES):return float(data)# This is kept for backwards compatibility even# if 'INTEGER_TYPES' isn't used by default anymore.elif INTEGER_TYPES and isinstance(data, INTEGER_TYPES):return int(data)# Special cases for numpy and pandas types# These are expensive to import so we try them last.serialized, value = _attempt_serialize_numpy_or_pandas(data)if serialized:return valueraise TypeError(f"Unable to serialize {data!r} (type: {type(data)})")class EsHandler(logging.Handler):def __init__(self, url, index, batch_size=100, flush_interval=6, extra_fields=None):super().__init__()self.default_raise = Falseself.index = indexself.batch_size = batch_sizeself.flush_interval = flush_intervalself.extra_fields = extra_fields if isinstance(extra_fields, dict) else {}self._buffer = []self.url = urlself._timer = Noneself.serializer = EsJsonSerializer()@staticmethoddef __get_es_datetime_str(timestamp):current_date = datetime.utcfromtimestamp(timestamp)return "{0!s}.{1:03d}Z".format(current_date.strftime('%Y-%m-%dT%H:%M:%S'), int(current_date.microsecond / 1000))def emit(self, record: logging.LogRecord) -> None:self.format(record)rec = self.extra_fields.copy()for key, value in record.__dict__.items():rec[key] = "" if value is None else valuerec["timestamp"] = self.__get_es_datetime_str(record.created)self._buffer.append(rec)# 如果缓存已满，直接刷新if len(self._buffer) >= self.batch_size:self.flush()else:self.schedule_flush()def schedule_flush(self):if self._timer is None:self._timer = Timer(self.flush_interval, self.flush)self._timer.daemon = Trueself._timer.start()def flush(self):# 如果缓存中有日志，则批量写入if self._timer is not None and self._timer.is_alive():self._timer.cancel()self._timer = Noneif self._buffer:try:logs_buffer = self._bufferself._buffer = []self.bulk_data(logs_buffer)except Exception as exception:if self.default_raise:raise exceptiondef bulk_data(self, logs_buffer):print("批量写入 Elasticsearch")request_body = ""for log_record in logs_buffer:actions = {"index": {'_index': self.index, "_id": str(uuid4())}}data = json.dumps(actions)request_body += datarequest_body += "\n"request_body += EsJsonSerializer().json_dumps(log_record)request_body += "\n"request_body += "\n"headers = {'content-type': 'application/json'}requests.post(self.url, data=request_body, headers=headers)def close(self):if self._timer is not None:self.flush()self._timer = None

四、日志格式介绍

字段作用
%(levelname)s日志级别（如 DEBUG、INFO、WARNING、ERROR、CRITICAL），用于标识日志的严重程度。
%(asctime)s日志的时间戳，默认格式为 YYYY-MM-DD HH:MM:SS,mmm（可自定义 datefmt）。
%(process)d进程 ID，表示当前日志属于哪个进程（适用于多进程应用）。
%(filename)s记录日志的 Python 文件名（不包含路径）。
%(name)s记录日志的 Logger 名称，通常是 logging.getLogger(name) 设定的名称。
%(lineno)d代码中的行号，表示日志记录发生的确切位置。
%(module)s模块名称，即 filename 去掉 .py 的部分。
%(message)s实际的日志信息，即 logger.info(“内容”) 传入的内容。
%(levelno)s日志级别的数值（如 DEBUG=10, INFO=20, WARNING=30, ERROR=40, CRITICAL=50）
%(pathname)s记录日志的 Python 文件完整路径。
%(funcName)s记录日志所在的函数名。
%(thread)d线程 ID（适用于多线程应用）。
%(threadName)s线程名称。
%(processName)s记录日志时距 logging 模块加载后的毫秒数。
%(msecs)d记录日志的时间戳的毫秒部分。
# 示例
LOG_FORMAT = '%(levelname)s - %(asctime)s - [PID: %(process)d] [Thread: %(threadName)s] - %(pathname)s - %(funcName)s - Line: %(lineno)d - %(message)s'

五、完整的代码

import logging
import sys
import os
import json
import socket
import time
import uuid
from datetime import date, datetime
from decimal import Decimal
from typing import Any, ClassVar, Mapping, Optional, Tuplefrom threading import Timer
from multiprocessing import Lock
import requests
from uuid import uuid4LOG_ENABLED = True  # 是否开启日志
LOG_TO_CONSOLE = True  # 是否输出到控制台
CONSOLE_LEVEL = "INFO"  # 控制台的日志等级
LOG_TO_FILE = True  # 是否输出到文件
FILE_LEVEL = "INFO"  # 文件的日志等级
LOG_TO_ES = True  # 是否输出到 Elasticsearch
ES_LEVEL = "INFO"  # 输出到ES的日志等级LOG_LEVEL = 'DEBUG'  # 日志级别LOG_FORMAT = '%(levelname)s - %(asctime)s - process: %(process)d - %(filename)s - %(name)s - %(lineno)d - %(module)s - %(message)s'  # 每条日志输出格式ELASTIC_SEARCH_HOST = 'http://192.168.3.200:9200'  # Elasticsearch Host
ELASTIC_SEARCH_INDEX = 'python'  # Elasticsearch Index Name
APP_NAME = "demo-fastapi"
APP_ENVIRONMENT = 'dev'  # 运行环境，如测试环境还是生产环境
LOG_DIR = os.getcwd()class JsonSerializer:"""JSON serializer relying on the standard library json module."""mimetype: ClassVar[str] = "application/json"def default(self, data: Any) -> Any:if isinstance(data, date):return data.isoformat()elif isinstance(data, uuid.UUID):return str(data)elif isinstance(data, Decimal):return float(data)raise Exception(f"Unable to serialize to JSON: {data!r} (type: {type(data).__name__})", )def json_dumps(self, data: dict) -> str:return json.dumps(data, default=self.default, ensure_ascii=False, separators=(",", ":"))def json_loads(self, data: str) -> dict:return json.loads(data)def loads(self, data: str) -> dict:return self.json_loads(data)def dumps(self, data: dict) -> str:# The body is already encoded to bytes# so we forward the request body along.if isinstance(data, str):return datareturn self.json_dumps(data)def _attempt_serialize_numpy(data: Any) -> Tuple[bool, Any]:try:import numpy as npif isinstance(data,(np.int_,np.intc,np.int8,np.int16,np.int32,np.int64,np.uint8,np.uint16,np.uint32,np.uint64,),):return True, int(data)elif isinstance(data,(np.float16,np.float32,np.float64,),):return True, float(data)elif isinstance(data, np.bool_):return True, bool(data)elif isinstance(data, np.datetime64):return True, data.item().isoformat()elif isinstance(data, np.ndarray):return True, data.tolist()except ImportError:# Since we failed to import 'numpy' we don't want to try again.return False, Nonereturn False, Nonedef _attempt_serialize_pandas(data: Any) -> Tuple[bool, Any]:try:import pandas as pdif isinstance(data, (pd.Series, pd.Categorical)):return True, data.tolist()elif isinstance(data, pd.Timestamp) and data is not getattr(pd, "NaT", None):return True, data.isoformat()elif data is getattr(pd, "NA", None):return True, Noneexcept ImportError:# Since we failed to import 'pandas' we don't want to try again.return False, Nonereturn False, Nonedef _attempt_serialize_numpy_or_pandas(data: Any) -> Tuple[bool, Any]:serialized, value = _attempt_serialize_numpy(data)if serialized:return serialized, valueserialized, value = _attempt_serialize_pandas(data)if serialized:return serialized, valuereturn False, NoneTIME_TYPES = (date, datetime)
FLOAT_TYPES = (Decimal,)
INTEGER_TYPES = ()class EsJsonSerializer(JsonSerializer):mimetype: ClassVar[str] = "application/json"def default(self, data: Any) -> Any:if isinstance(data, TIME_TYPES):# Little hack to avoid importing pandas but to not# return 'NaT' string for pd.NaT as that's not a valid# Elasticsearch date.formatted_data = data.isoformat()if formatted_data != "NaT":return formatted_dataif isinstance(data, uuid.UUID):return str(data)elif isinstance(data, FLOAT_TYPES):return float(data)# This is kept for backwards compatibility even# if 'INTEGER_TYPES' isn't used by default anymore.elif INTEGER_TYPES and isinstance(data, INTEGER_TYPES):return int(data)# Special cases for numpy and pandas types# These are expensive to import so we try them last.serialized, value = _attempt_serialize_numpy_or_pandas(data)if serialized:return valueraise TypeError(f"Unable to serialize {data!r} (type: {type(data)})")class EsHandler(logging.Handler):def __init__(self, url, index, batch_size=100, flush_interval=6, extra_fields=None):super().__init__()self.default_raise = Falseself.index = indexself.batch_size = batch_sizeself.flush_interval = flush_intervalself.extra_fields = extra_fields if isinstance(extra_fields, dict) else {}self._buffer = []self.url = urlself._timer = Noneself.serializer = EsJsonSerializer()@staticmethoddef __get_es_datetime_str(timestamp):current_date = datetime.utcfromtimestamp(timestamp)return "{0!s}.{1:03d}Z".format(current_date.strftime('%Y-%m-%dT%H:%M:%S'), int(current_date.microsecond / 1000))def emit(self, record: logging.LogRecord) -> None:self.format(record)rec = self.extra_fields.copy()for key, value in record.__dict__.items():rec[key] = "" if value is None else valuerec["timestamp"] = self.__get_es_datetime_str(record.created)self._buffer.append(rec)# 如果缓存已满，直接刷新if len(self._buffer) >= self.batch_size:self.flush()else:self.schedule_flush()def schedule_flush(self):if self._timer is None:self._timer = Timer(self.flush_interval, self.flush)self._timer.daemon = Trueself._timer.start()def flush(self):# 如果缓存中有日志，则批量写入if self._timer is not None and self._timer.is_alive():self._timer.cancel()self._timer = Noneif self._buffer:try:logs_buffer = self._bufferself._buffer = []self.bulk_data(logs_buffer)except Exception as exception:if self.default_raise:raise exceptiondef bulk_data(self, logs_buffer):print("批量写入 Elasticsearch")request_body = ""for log_record in logs_buffer:actions = {"index": {'_index': self.index, "_id": str(uuid4())}}data = json.dumps(actions)request_body += datarequest_body += "\n"request_body += EsJsonSerializer().json_dumps(log_record)request_body += "\n"request_body += "\n"headers = {'content-type': 'application/json'}requests.post(self.url, data=request_body, headers=headers)def close(self):if self._timer is not None:self.flush()self._timer = Noneclass Logger:_instance = None_lock = Lock()def __init__(self, ):self.logger = logging.getLogger("__file__")self.logger.setLevel(LOG_LEVEL)self.add_stream_handle()self.add_file_handle()self.add_es_handle()def add_stream_handle(self):if LOG_TO_CONSOLE and LOG_ENABLED:handler = logging.StreamHandler(sys.stdout)handler.setLevel(level=CONSOLE_LEVEL)handler.setFormatter(logging.Formatter(LOG_FORMAT))self.logger.addHandler(handler)def add_file_handle(self):if LOG_ENABLED and LOG_TO_FILE:logs_dir = os.path.join(LOG_DIR, "logs")os.makedirs(logs_dir, exist_ok=True)file = os.path.join(logs_dir, f'{datetime.now().strftime("%Y%m%d_%H%M%S")}.log')handler = logging.FileHandler(file, encoding="utf-8")handler.setLevel(level=FILE_LEVEL)handler.setFormatter(logging.Formatter(LOG_FORMAT))self.logger.addHandler(handler)def add_es_handle(self):if LOG_ENABLED and LOG_TO_ES:# 添加 CMRESHandlerurl = f"{ELASTIC_SEARCH_HOST}/_bulk"extra_fields = {"app_name": APP_NAME, "app_env": APP_ENVIRONMENT, 'host': socket.gethostname(),'host_ip': socket.gethostbyname(socket.gethostname())}es_handler = EsHandler(url, index=ELASTIC_SEARCH_INDEX, extra_fields=extra_fields)es_handler.setLevel(level=LOG_LEVEL)formatter = logging.Formatter(LOG_FORMAT)es_handler.setFormatter(formatter)self.logger.addHandler(es_handler)@classmethoddef get_logger(cls):cls._lock.acquire()if cls._instance is None:cls._instance = Logger().loggercls._lock.release()return cls._instanceif __name__ == '__main__':logger = Logger.get_logger()logger.info("INFO")logger.warning("WARNING")logger.error("ERROR")time.sleep(100)

结语

掌握 Python 的 logging 模块，能让你的日志管理更加高效，无论是简单的控制台调试，还是日志文件持久化存储，甚至是对接 ELK 进行集中管理，都是开发者必备的技能。

http://www.dtcms.com/wzjs/374246.html

相关文章：

  • 音乐网站建设成本seo81
  • 国外用python做的网站谁能给我个网址
  • 单页网站订单系统怎么改邮箱西安网站seo价格
  • 杭州哪家公司做网站比较好崇左网站建设
  • 想做网站怎么做seo是什么?
  • b2b网站的主要功能和作用是什么搜索引擎优化实训心得
  • wordpress导出主题代码seo排名点击器
  • 人才网站 建设好 窗口神马移动排名优化
  • 网站建设西安台州seo
  • wordpress 评论贴图seo怎么做教程
  • asp网站开发工具神器市场营销
  • 为什么做电影网站没有流量站长工具箱
  • 网站备案幕布照片无经验能做sem专员
  • 怎么申请免费国内免费网站武汉seo网站优化排名
  • 做企业网站时需要注意哪些地方品牌推广策略与方式
  • 怎样使用wordpress网络seo是什么
  • 中文网站怎么做英文版镇江关键字优化公司
  • 福州手游网站建设网店代运营十大排名
  • 做医学网站广州发布紧急通知
  • 黑山网站制作公司发广告去哪个平台
  • 手机网站版面设计seo网站优化专员
  • 建设银行官方网站手机版网络营销试题库及答案
  • 做英文兼职的网站网络营销做得好的品牌
  • 网站建设参考的文献建站是什么意思
  • asp做网站搜索网站快速建站
  • 网站建设属于哪个经营范围特大新闻凌晨刚刚发生
  • 网站建设公司上海平台优化
  • 河南中国建设银行官网站在线超级外链工具
  • 延安网站建设目录型搜索引擎有哪些
  • 苏州外贸网站设计网站推广软件免费版