五十岚 ▪ 寄

文件基础

五十岚2020年5月3日
大约 7 分钟

上下文管理器 可能与子程序（subroutine）本身一样重要，它会做 事前/后 的清理操作

文件

1. 文件基础

Unix 理念：一切皆文件

  • 包括操作系统，所有的一切都是文件
  • 因此对于所有，无非涉及：打开 、操作 、 关闭 这 3 个动作，即文件操作的雏形

1.1 文件操作

Python 操作文件使用 open()open in new window

open(file, mode='r', buffering=- 1, encoding=None, errors=None, newline=None, closefd=True, opener=None)

打开 文件 file，并返回对应的 文件对象，有如下三种

  1. 原始 二进制文件open in new window

  2. 缓冲 二进制文件open in new window

  3. 文本文件open in new window

# 假设有文本文件 1.txt
filename = "1.txt"

f = open(filename, mode = "r", encoding="utf-8")

  • mode: str 可指定如下，可 任意组合

    mode 字符含意
    'r'读取（默认），以文本模式打开并读取
    'w'写入，存在则清空等待写入，无则创建
    'x'排它性创建，如果文件已存在则失败
    'a'打开文件用于写入，如果文件存在则在末尾追加
    'b'用二进制的模式打开文件，返回 bytes，不解码
    't'用文本模式（默认）打开，返回 str，指定解码
    '+'打开用于更新（读取与写入）
    • r+ 最常用，即可读写模式
    • w+ 写读模式，与 r+ 区别在，会 先执行 一次 .trancate() 操作
    • a+ 追加写模式，用的少

  • encoding: str 指定编码方式

  • f = open(...) 获取文件对象（句柄 / 文件描述符）

常用方法

此时返回的文件对象 f 具有如下常用方法

f.read(size=5)				# 读取	读取文件的 5个字符(中英同)，省略、负数，则是读所有

f.write("\t 2333\n")		# 写入	往文件中写入字符串 "\t 2333\n"

f.close() 					# 关闭	关闭文件句柄，若不关闭，python默认机制也会关闭，但不可靠
  • 执行 write() 操作时，会将写入的字符串暂存到缓冲区，只有执行 close() 操作后，才会落盘
f.readline() 				# 读取一行	从文件中读取单行数据，会保留换行符 `\n`

f.readlines() 				# 读取所有	将所有内容返回为列表（1行 == 1个元素)

  • 使用 readlines() 很快，但有内存开销，因此对于大容量文件，常用无内存开销的 迭代器

    for line in f:
	print(line, end='')

f.tell() 					# 取位置	返回游标整数，给出文件对象在文件中的当前位置

f.seek(offset, whence) 		# 改位置	修改文件对象的游标

  • 通常用于客户端向服务端发送文件，断网后的 断点续传，tell() 返回游标，再用 seek(index) 指定断点，继续传输

  • seek(0) 表示文件头

  • 通过向 whence （参考点），添加 offset（偏移） 计算位置， whence 值如下

    • 默认值为 0 ，表示从文件开头计算
    • 指定 1 ，表示使用当前文件位置
    • 指定 2， 表示从文件末尾计算
    f = open('1.txt', 'rb+')
f.write(b'0123456789abcdef')

f.seek(5)		# 将文件对象的游标指向 第6顺位
## 5
f.tell()		# 返回游标
## 5
f.read(1)		# 读取 文件对象的第1个字节
## b'5'

f.seek(-3, 2)	# 尾部参考点, 将文件对象的游标指向 第3逆位
## 13
f.read(1)
## b'd'
不常用方法
f.flush()			# 刷新	刷新文件内部缓冲，将缓冲区的文件立即写入文件，.close()会自动刷新缓冲区
    # - 常用来提前落盘，防止断电内存丢数据
    # - 写命令行显示的 进度条 sys.stdout.write("*")  sys.stdout.flush()
	# - .close()会自动刷新缓冲区


f.trancate()		# 截断	从指定长度进行截断，此时只能获取截断前的字符
	# - 空参为0，此时截取 0个字符串，即清空
    # - "w" 则是默认执行一次


f.flieno()			# 返回一个整型的文件描述符 (文件在内存中 唯一创建 的文件编号)
f.isatty()			# 检测文件对象 是否连接到一个 终端设备 (如打印机)

f.readable()		# 是否可读
f.writeable()		# 是否可写
f.seekable()		# 是否可追踪

2. 上下文管理器

Python 通过 with 语句open in new window，支持 上下文管理器 所定义的，运行时上下文

  • 上下文管理器，是个实现了上下文管理协议（context management protocol）的对象，实现它需要__enter__ / __exit__ 俩方法，本质是装饰器
  • 它允许用户 自定义类 来 定义运行时的上下文
  • 它会在 语句体被执行前（代码执行前） 进入该上下文，并在 语句执行完毕时 （代码执行后） 退出该上下文

2.1 管理协议

实现管理协议，需要实现如下 两个方法

contextmanager.__enter__()

调用 with 时，会通过该方法 进入运行时上下文，若 with 语句中有 as 子句，返回值 会绑定给 as 后的变量

with open("/etc/hosts", "r") as file:
    dir(file)

## [..., '__enter__', '__exit__', ...]
  • 文件对象会从 __enter__() 返回当前对象（自己），让 open() 作为 with 语句的上下文表达式
  • 返回的文件对象会赋值给 file 变量，该对象的文件类实现了上下文管理协议

contextmanager.__exit__(exc_type, exc_val, exc_tb) -> bool

退出运行时的上下文，若运行时发生异常，则会退出上下文管理器，并 调用此方法，它返回一个 bool 标识，来检测是否有需要处理的异常，退出时参数如下

  • exc_type 异常类型
  • exc_val 异常值
  • exc_tb 异常追踪信息

若未发生异常，则三个参数都为 None

2.2 with 语句

为了 支持上下文管理器存在的，使用上下文管理协议的方法包裹一个代码块，可为 try - except - finally 提供方便的封装，写法如下

with EXPRESSION  as TARGET:
	BLOCK
  • with 和 as 是 关键词， 其中 as 可选
  • EXPRESSION 是 上下文表达式
  • TARGET 是赋值的目标变量
  • BLOCK 被包裹的代码块

语义上，等同于

manager = (EXPRESSION)						# 上下文管理器的表达式
enter = type(manager).__enter__				# 通过 __enter__ 初始化 exit
exit = type(manager).__exit__  				# 通过 __exit__ 初始化 exit
value = enter(manager)						# 调用 enter() 执行上下文
hit_except = False

try:
    TARGET = value							# 如果有 "as TARGET"
    BLOCK

except:
    hit_except = True
    if not exit(manager, *sys.exc_info()):	# 捕获异常则执行 exit()， 尝试退出
        raise
        # 若__exit__()返回 False，异常将被传播（无法处理）;若返回 True，异常将被终止（正常解决）

finally:
    if not hit_except:
        exit(manager, None, None, None)		# 无异常发生，则执行 exit() 正常退出

3.1 后可以组合 多种 表达式

with A() as a, B() as b:
    BLOCK

# 等价于
with A() as a:
    with B() as b:
        BLOCK

2.3 自定义上下文管理器

当 with 语句不使用 as 关键词，会直接返回当前对象

class ContextManager(object):
    def __init__(self):
        print("__init__()")

    def __enter__(self):
        print("__enter__()")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"__exit__() - param: {exc_type, exc_val, exc_tb}")

with ContextManager():
    print("被自定义上下文 包裹的代码块")

## 输出结果
# __init__()
# __enter__()
# 被自定义上下文 包裹的代码块
# __exit__() - param: (None, None, None)

对于异常处理

class InnerContext(object):
    def __init__(self, obj):
        print(f"InnerContext.__init__({obj})")

    def do_something(self):
        print("InnerContext.do_something()")

    def __del__(self):
        print("InnerContext.__del__()")


class ContextManager(object):
    def __init__(self, flag):
        print(f"ContextManager.__init__(flag)")
        self.flag = flag

    def __enter__(self):
        print("ContextManager.__enter__()")
        return InnerContext(self)

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"ContextManager.__exit__({exc_type}, {exc_val}, {exc_tb})")
        return self.flag

with ContextManager(True) as obj:
    print(f"obj = {obj}")
    obj.do_something()
    raise RuntimeError('error message handled')

## 输出结果
# ContextManager.__init__(flag)
# ContextManager.__enter__()
# InnerContext.__init__(<__main__.ContextManager object at 0x000001C3676978E0>)
# InnerContext.__del__()
# obj = <__main__.InnerContext object at 0x000001C367703820>
# InnerContext.do_something()
# ContextManager.__exit__(<class 'RuntimeError'>, error message handled, <traceback object at 0x000001C3677B4BC0>)
----------------------------------------------------------------------------------------

with ContextManager(False) as obj:
    raise RuntimeError('error message propagated')	# 异常传播

## 输出结果
# ContextManager.__init__(flag)
# ContextManager.__enter__()
# InnerContext.__init__(<__main__.ContextManager object at 0x000001C3677D67F0>)
# InnerContext.__del__()
# ContextManager.__exit__(<class 'RuntimeError'>, error message propagated, <traceback object at 0x000001C367755900>)
---------------------------------------------------------------------------
# RuntimeError                              Traceback (most recent call last)
# <ipython-input-125-02e252020534> in <module>
#       1 with ContextManager(False) as obj:
# ----> 2     raise RuntimeError('error message propagated')
#       3

# RuntimeError: error message propagated
  • 对于 __exit__() 的返回值为 True ，表示正常退出，不会继续 raise 传播

3. 进阶

上次编辑于: 2022/9/20 06:19:59
默认页脚
Copyright © 2023 五十岚