這篇文章主要介紹“Python的logger怎么配置”,在日常操作中,相信很多人在Python的logger怎么配置問題上存在疑惑,小編查閱了各式資料,整理出簡單好用的操作方法,希望對大家解答”Python的logger怎么配置”的疑惑有所幫助!接下來,請跟著小編一起來學習吧!
成都創(chuàng)新互聯(lián)專注為客戶提供全方位的互聯(lián)網(wǎng)綜合服務,包含不限于成都做網(wǎng)站、網(wǎng)站建設、布爾津網(wǎng)絡推廣、微信小程序定制開發(fā)、布爾津網(wǎng)絡營銷、布爾津企業(yè)策劃、布爾津品牌公關(guān)、搜索引擎seo、人物專訪、企業(yè)宣傳片、企業(yè)代運營等,從售前售中售后,我們都將竭誠為您服務,您的肯定,是我們最大的嘉獎;成都創(chuàng)新互聯(lián)為所有大學生創(chuàng)業(yè)者提供布爾津建站搭建服務,24小時服務熱線:18982081108,官方網(wǎng)址:vcdvsql.cn
傳遞的數(shù)據(jù)結(jié)構(gòu)如何考慮(是否對調(diào)用方有先驗知識的要求,比如返回一個 Tuple,則需要用戶了解 tuple 中元素的順序,這樣情況是否應該進行封裝;),數(shù)據(jù)結(jié)構(gòu)定義清楚了,很多東西也就清楚了。
如何操作數(shù)據(jù)庫(可以學習 sqlalchemy,包括 core 和 orm 兩種 api)
異常如何處理(異常應該分開捕獲 — 可以清楚的知道什么情況下導致的,異常之后應該打印日志說明出現(xiàn)什么問題,如果情況惡劣需要進行異常再次拋出或者報警)
所有獲取資源的地方都應該做 check(a. 沒有獲取到會怎么辦;b.獲取到異常的怎么辦)
所有操作資源的地方都應該檢查是否操作成功
每個函數(shù)都應該簡短,如果函數(shù)過長應該進行拆分(有個建議值,函數(shù)包含的行數(shù)應該在 20-30 行之間,具體按照這個規(guī)范做過一次之后就會發(fā)現(xiàn)這樣真好)
使用 class 之后,考慮重構(gòu) __str__ 函數(shù),用戶打印輸出(如果不實現(xiàn) __str__,會調(diào)用 __repr__ ),如果對象放到 collection 中之后,需要實現(xiàn) __repr__ 函數(shù),用于打印整個 collection 的時候,直觀顯示
如果有些資源會發(fā)生變化,可以單獨抽取出來,做成函數(shù),這樣后續(xù)調(diào)用就可以不用改變了
附上一份 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 是為了在函數(shù)間傳遞時,不需要使用方了解屬性的具體順序而寫的,也可以放到一個單獨的模塊中 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() # 因為下面的資源可能發(fā)生變化,而且可能包含一些具體的邏輯,因此單獨抽取出來,獨立成一個函數(shù) 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
到此,關(guān)于“Python的logger怎么配置”的學習就結(jié)束了,希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習,快去試試吧!若想繼續(xù)學習更多相關(guān)知識,請繼續(xù)關(guān)注創(chuàng)新互聯(lián)網(wǎng)站,小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>
網(wǎng)頁標題:Python的logger怎么配置
路徑分享:http://vcdvsql.cn/article48/pehphp.html
成都網(wǎng)站建設公司_創(chuàng)新互聯(lián),為您提供、做網(wǎng)站、服務器托管、品牌網(wǎng)站制作、品牌網(wǎng)站建設、關(guān)鍵詞優(yōu)化
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時需注明來源: 創(chuàng)新互聯(lián)