背景
作为一个有追求的 Python 程序猿,我一直在寻找更加简洁的实现方式。这个过程就有点像小时候在河边捡石子,不断地发现新的石子放进口袋;到家之后又弃之十之八九,最终得到那颗“明珠”。
Python 的学习过程也差不多,最开始可能是发现了一个新的写法,然后自己去官方文档里面补充大量的背景知识,最后又忘记了那些不常用的部分。最终剩下的这么一星半点应该可以算得上是 “精华” 部分了。
在 Python 社区混了差不多 10 年,“精华” 也是沉淀了不少,如果要我说一个印象最为深刻的,我想应该是上下文管理协议。
天不生我上下文,万古 try-catch 长如夜
没有上下文件管理之前,我们只能依赖于 try-catch-finally 这种异常处理结构,当然 Python 中并没有使用 catch 而是用 except 这个关键字。就拿打开一个文件来说了,样板代码一大堆。
代码语言:javascript复制#!/usr/bin/env python3
import logging
logging.basicConfig(level=logging.INFO)
def main():
"""向 /tmp/a.txt 文件中写入 "hello world ."
"""
file_object = None
try:
logging.info("open file in try block. ")
# 打开文件
file_object = open("/tmp/a.txt", 'w')
file_object.write("hello world .n")
except Exception as err:
pass
finally:
# 关闭文件回收句柄
logging.info("close file in finally block .")
if file_object is not None:
file_object.close()
main()
使用 with 上下文管理协议之后,我们可以去掉所有的样板代码,整体的代码行数就降了下来。
代码语言:javascript复制#!/usr/bin/env python3
import logging
logging.basicConfig(level=logging.INFO)
def main():
"""向 /tmp/a.txt 文件中写入 "hello world ."
"""
with open("/tmp/a.txt",'w') as file_object:
file_object.write("hello world .n")
main()
with 背后的原理
with 语句背后的原理就是上下文管理协议,这个协议约定了进入上下文之前要调用对象的 __enter__ 方法,在退出的时候要调用对象的 __exit__ 方法;我们可以在 __enter__ 方法里面实现 try 部分的逻辑,而在 __exit__ 方法里面实现 finally 的逻辑。
在写正式的代码之前我们可以先搭好协议的架子。让我们从零开始实现一个支持 with 的对象吧。
代码语言:javascript复制#!/usr/bin/env python3
import logging
logging.basicConfig(level=logging.INFO)
class MFile(object):
file_path:str = None
_file_object = None
def __init__(self, file_path:str=None):
self.file_path = file_path
def __enter__(self):
logging.info("__enter__ function called .")
self._file_object = open(self.file_path)
def __exit__(self, *args, **kwargs):
logging.info("__exit__ function called .")
if self._file_object is not None:
self._file_object.close()
def main():
with MFile("/tmp/a.txt") as f:
pass
main()
运行效果是这样的。
代码语言:javascript复制python3 main.py
INFO:root:__enter__ function called .
INFO:root:__exit__ function called .
可以看到现在已经支持上下文件管理协议了。事实上我们只要再加一点代码就能和内置的 file 对象一样了,下面我们让它支持 write 函数吧。
代码语言:javascript复制#!/usr/bin/env python3
import logging
logging.basicConfig(level=logging.INFO)
class MFile(object):
file_path:str = None
_file_object = None
def __init__(self, file_path:str=None):
self.file_path = file_path
def __enter__(self):
logging.info("__enter__ function called .")
self._file_object = open(self.file_path, 'w')
return self
def __exit__(self, exc_type, exc_val, exc_tb):
logging.info("__exit__ function called .")
if self._file_object is not None:
self._file_object.close()
def write(self, *args, **kwargs):
"""让 MFile 支持 write
"""
self._file_object.write(*args, **kwargs)
def main():
with MFile("/tmp/a.txt") as f:
f.write("just a test str . n")
main()
查看运行的效果。
代码语言:javascript复制python3 main.py
INFO:root:__enter__ function called .
INFO:root:__exit__ function called .
cat /tmp/a.txt
just a test str .
简化 with 上下文件管理协议的实现
要想简化 with 上下管理协议的实现,我还要跳出来从更高的层审视 __enter__ && __exit__ 的执行时机。
1. 最开始是执行的 __entrer__
2. 执行其他的业务逻辑
3. 最后是执行 __exit__
可以看到执行流程被分成了三段,__enter__ 和 __exit__ 之间是不连续的。这个就有点像上下文对象执行完 __enter__ 就 “睡” 了,直到要执行 __exit__ 的时候这才 “醒” 过来,这个行为不就是 Python 中的生成器吗!!!
是的这里用生成器可以让我们的代码再少写许多行,下面演示一下。
代码语言:javascript复制#!/usr/bin/env python3
import logging
import contextlib
logging.basicConfig(level=logging.INFO)
@contextlib.contextmanager
def MFile(file_path:str=None):
# __enter__
logging.info("__enter__ function called .")
file_object = open(file_path, 'w')
yield file_object
# __exit__
logging.info("__exit__ function called .")
file_object.close()
def main():
with MFile("/tmp/a.txt") as f:
f.write("just a test str . n")
main()
查看运行效果
代码语言:javascript复制python3 main.py
INFO:root:__enter__ function called .
INFO:root:__exit__ function called .
cat /tmp/a.txt
just a test str .
可以看到输出的结果和我期望的一样,总的来讲 Python 是一门非常有意思的语言。主要表现在当我们知道得越多,我们写出来的代码就越是短。
项目实践
这个技巧我在项目还是经常用的,最多的一个应该算是 sudo 了。平时我的程序用的是普通用户权限运行,当它需要用到高权限时,我就让它 sudo 到 root 去,用完成之后再回到普通用户。
代码语言:javascript复制import os
import contextlib
from threading import RLock
_user_sudo_lock = RLock()
@contextlib.contextmanager
def sudo(message="sudo"):
"""临时升级权限到 root .
"""
# 对于权限这个临界区的访问要串行化
with _user_sudo_lock as lk:
# 得到当前进程的 euid
old_euid = os.geteuid()
# 提升权限到 root
os.seteuid(0)
yield message
# 恢复到普通权限
os.seteuid(old_euid)
github 路径: https://github.com/Neeky/dbm-agent/blob/master/dbma/bil/sudos.py