重庆分公司,新征程启航

为企业提供网站建设、域名注册、服务器等服务

Python的logger怎么配置

这篇文章主要介绍“Python的logger怎么配置”,在日常操作中,相信很多人在Python的logger怎么配置问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”Python的logger怎么配置”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!

成都创新互联专注为客户提供全方位的互联网综合服务,包含不限于成都做网站、网站建设、布尔津网络推广、微信小程序定制开发、布尔津网络营销、布尔津企业策划、布尔津品牌公关、搜索引擎seo、人物专访、企业宣传片、企业代运营等,从售前售中售后,我们都将竭诚为您服务,您的肯定,是我们最大的嘉奖;成都创新互联为所有大学生创业者提供布尔津建站搭建服务,24小时服务热线:18982081108,官方网址:www.cdcxhl.com

传递的数据结构如何考虑(是否对调用方有先验知识的要求,比如返回一个 Tuple,则需要用户了解 tuple 中元素的顺序,这样情况是否应该进行封装;),数据结构定义清楚了,很多东西也就清楚了。

如何操作数据库(可以学习 sqlalchemy,包括 core 和 orm 两种 api)

异常如何处理(异常应该分开捕获 — 可以清楚的知道什么情况下导致的,异常之后应该打印日志说明出现什么问题,如果情况恶劣需要进行异常再次抛出或者报警)

所有获取资源的地方都应该做 check(a. 没有获取到会怎么办;b.获取到异常的怎么办)

所有操作资源的地方都应该检查是否操作成功

每个函数都应该简短,如果函数过长应该进行拆分(有个建议值,函数包含的行数应该在 20-30 行之间,具体按照这个规范做过一次之后就会发现这样真好)

使用 class 之后,考虑重构 __str__ 函数,用户打印输出(如果不实现 __str__,会调用 __repr__ ),如果对象放到 collection 中之后,需要实现 __repr__ 函数,用于打印整个 collection 的时候,直观显示

如果有些资源会发生变化,可以单独抽取出来,做成函数,这样后续调用就可以不用改变了

附上一份 Python2.7 代码(将一些私有的东西进行了修改)

# -*- coding:utf-8 -*-   from sqlalchemy import create_engine import logging from logging.config import fileConfig import requests import Clinet # 私有的模块   fileConfig("logging_config.ini") logger = logging.getLogger("killduplicatedjob")   #配置可以单独放到一个模块中 DB_USER = "xxxxxxx" DB_PASSWORD = "xxxxxxxx" DB_PORT = 111111 DB_HOST_PORT = "xxxxxxxxxx" DB_DATA_BASE = "xxxxxxxxxxx"   REST_API_URL = "http://sample.com"   engine = create_engine("MySQL://%s:%s@%s:%s/%s" % (DB_USER, DB_PASSWORD, DB_HOST_PORT, DB_PORT, DB_DATA_BASE))   # 这个 class 是为了在函数间传递时,不需要使用方了解属性的具体顺序而写的,也可以放到一个单独的模块中 class DuplicatedJobs(object):  def __init__(self, app_id, app_name, user):  self.app_id = app_id  self.app_name = app_name  self.user = user   def __repr__(self):  return '[appid:%s, app_name:%s, user:%s]' % (self.app_id, self.app_name, self.user)   def find_duplicated_jobs():  logger.info("starting find duplicated jobs")  (running_apps, app_name_to_user) = get_all_running_jobs()  all_apps_on_yarn = get_apps_from_yarn_with_queue(get_resource_queue())   duplicated_jobs = []  for app in all_apps_on_yarn:  (app_id, app_name) = app   if app_id not in running_apps:  if not app_name.startswith("test"):  logger.info("find a duplicated job, prefixed_name[%s] with appid[%s]" % (app_name, app_id))  user = app_name_to_user[app_name]  duplicated_jobs.append(DuplicatedJobs(app_id, app_name, user))  else:  logger.info("Job[%s] is a test job, would not kill it" % app_name)   logger.info("Find duplicated jobs [%s]" % duplicated_jobs)   return duplicated_jobs   def get_apps_from_yarn_with_queue(queue):  param = {"queue": queue}  r = requests.get(REST_API_URL, params=param)  apps_on_yarn = []  try:  jobs = r.json().get("apps")  app_list = jobs.get("app", [])  for app in app_list:  app_id = app.get("id")  name = app.get("name")  apps_on_yarn.append((app_id, name))   except Exception as e: #Exception ***进行单独的分开,针对每一种 Exception 进行不同的处理  logger.error("Get apps from Yarn Error, message[%s]" % e.message)   logger.info("Fetch all apps from Yarn [%s]" % apps_on_yarn)   return apps_on_yarn   def get_all_running_jobs():  job_infos = get_result_from_mysql("select * from xxxx where xx=yy")   app_ids = []  app_name_to_user = {}  for (topology_id, topology_name) in job_infos:  status_set = get_result_from_mysql("select * from xxxx where xx=yy")  application_id = status_set[0][0]  if "" != application_id:  configed_resource_queue = get_result_from_mysql(  "select * from xxxx where xx=yy")  app_ids.append(application_id)  app_name_to_user[topology_name] = configed_resource_queue[0][0].split(".")[1]   logger.info("All running jobs appids[%s] topology_name2user[%s]" % (app_ids, app_name_to_user))  return app_ids, app_name_to_user   def kill_duplicated_jobs(duplicated_jobs):  for job in duplicated_jobs:  app_id = job.app_id  app_name = job.app_name  user = job.user  logger.info("try to kill job[%s] with appid[%s] for user[%s]" % (app_name, app_id, user))  try:  Client.kill_job(app_id, user)  logger.info("Job[%s] with appid[%s] for user[%s] has been killed" % (app_name, app_id, user))  except Exception as e:  logger.error("Can't kill job[%s] with appid[%s] for user[%s]" % (app_name, app_id, user))   def get_result_from_mysql(sql):  a = engine.execute(sql)  return a.fetchall()   # 因为下面的资源可能发生变化,而且可能包含一些具体的逻辑,因此单独抽取出来,独立成一个函数 def get_resource_queue():  return "xxxxxxxxxxxxx"   if __name__ == "__main__":  kill_duplicated_jobs(find_duplicated_jobs())

其中 logger 配置文件如下(对于 Python 的 logger,官方文档写的非常好,建议读一次,并且实践一次)

[loggers] keys=root, simpleLogger   [handlers] keys=consoleHandler, logger_handler   [formatters] keys=formatter   [logger_root] level=WARN handlers=consoleHandler   [logger_simpleLogger] level=INFO handlers=logger_handler propagate=0 qualname=killduplicatedjob   [handler_consoleHandler] class=StreamHandler level=WARN formatter=formatter args=(sys.stdout,)   [handler_logger_handler] class=logging.handlers.RotatingFileHandler level=INFO formatter=formatter args=("kill_duplicated_streaming.log", "a", 52428800, 3,)   [formatter_formatter] format=%(asctime)s %(name)-12s %(levelname)-5s %(message)s

到此,关于“Python的logger怎么配置”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注创新互联网站,小编会继续努力为大家带来更多实用的文章!


当前名称:Python的logger怎么配置
本文地址:http://cqcxhl.com/article/pehphp.html

其他资讯

在线咨询
服务热线
服务热线:028-86922220
TOP