Python 代码实践小结
最近写了较多的 Python 脚本,将最近自己写的脚本进行一个总结,其中有些是 Python 独有的,有些是所有程序设计中共有的:
考虑使用 Logger(logger 怎么配置,需要输出哪些信息 — 可以反向考虑,自己看到这个 logger 的时候想了解什么信息)
传递的数据结构如何考虑(是否对调用方有先验知识的要求,比如返回一个 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("Cant 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] "kill_duplicated_streaming.log", "a", 52428800, 3,) [formatter_formatter] format=%(asctime)s %(name)-12s %(levelname)-5s %(message)s
本文作者:佚名
来源:51CTO
Python编程从入门到实践-读书笔记(下) 基础知识重点摘录 在Python中,用引号括起的都是字符串,其中的引号可以是单引号,也可以是双引号。这种灵活性让你能够在字符串中包含引号和撇号:
Python编程从入门到实践-读书笔记(上) 基础知识重点摘录 在Python中,用引号括起的都是字符串,其中的引号可以是单引号,也可以是双引号。这种灵活性让你能够在字符串中包含引号和撇号:
相关文章
- python的for循环和while循环的一些代码
- 《大话设计模式》Python 版代码实现
- 【华为OD机试真题 python】玩牌高手 【2022 Q4 | 100分】
- 用Python做兼职接单,简直是爽到离谱
- python中matplotlib.pyplot的使用示例
- 153 python网络编程 - TCP客户端
- 34 python - 匿名函数
- python合并pdf
- 推荐系统-召回-负采样python代码
- 深度学习模型stacking模型融合python代码,看了你就会使
- 《Python高性能编程》——2.3 计算完整的Julia集合
- python和C++代码实现模拟动态指针时钟
- Python基础必掌握的利用Booleans优化代码巧妙方法
- k8s格式化apache日志正则表达式(python)
- Python代码库OpenCV之10圆检测circle detection 霍夫曼(含代码)
- Python 代码库之unicode 编码与字符串之间相互转换
- (数据科学学习手札13)K-medoids聚类算法原理简介&Python与R的实现
- Python 代码库之unicode 编码与字符串之间相互转换
- 《Python语言程序设计》——1.8 程序设计错误
- 《树莓派Python编程指南》——3.8 小结
- 一个简单的步骤让你的 Python 代码更干净
- 【华为OD机试真题 java、python、c++】优秀学员统计【2022 Q4 100分】(100%通过)
- Python版飞机大战拓展功能的开发
- 7、代码的完整性——调整数组顺序使奇数位在前偶数位在后(python版)
- websocket for python
- 华为OD机试 - 翻转单词顺序(Python) | 机试题+算法思路+考点+代码解析 【2023】
- 华为OD机试 - 网上商城优惠活动(一)(Python) | 机试题+算法思路+考点+代码解析 【2023】
- 华为OD机试 - 九宫格按键输入(Python) | 机试题+算法思路+考点+代码解析 【2023】
- 华为OD机试 -查找单入口空闲区域(Python) | 机试题+算法思路+考点+代码解析 【2023】
- 华为OD机试 - 单词倒序(Python) | 机试题+算法思路+考点+代码解析 【2023】
- 华为OD机试 - 异常的打卡记录(Python) | 机试题+算法思路+考点+代码解析 【2023】
- 华为OD机试 - 特异性双端队列(Python)| 真题+思路+考点+代码+岗位
- 利用Python自动在CDDIS上下载GNSS数据(观测值、广播星历,电离层格网数据)
- 学习笔记(13):Python网络编程&并发编程-解决粘包问题-终极版本
- 学习笔记(07):Python网络编程&并发编程-客户端与服务端代码bug修复