Python文件读写最佳实践:关键操作的异常处理
ztj100 2025-05-09 22:53 23 浏览 0 评论
在Python中进行文件操作时,合理的异常处理是保证程序健壮性的关键。以下是针对文件操作异常处理的全面指南。
一、为什么需要异常处理?
文件操作可能失败的常见原因:
- 文件不存在(FileNotFoundError)
- 权限不足(PermissionError)
- 磁盘已满(OSError)
- 编码问题(UnicodeDecodeError)
- 文件被占用(IOError)
- 硬件故障(OSError)
二、基础异常处理模式
1. 基本文件读取的异常处理
try:
with open('important.json', 'r', encoding='utf-8') as f:
data = json.load(f)
except FileNotFoundError:
print("错误:配置文件不存在,将使用默认配置")
data = default_config
except json.JSONDecodeError as e:
print(f"配置文件格式错误: {e}")
raise SystemExit(1) # 严重错误,终止程序
except Exception as e:
print(f"未知错误: {e}")
raise # 重新抛出未知异常
2. 文件写入的异常处理
try:
with open('output.log', 'a', encoding='utf-8') as f: # 使用追加模式
f.write(f"{datetime.now()}: 操作记录\n")
except PermissionError:
print("错误:没有写入权限,尝试备用位置")
write_to_alternate_location()
except OSError as e:
if e.errno == errno.ENOSPC:
print("错误:磁盘空间不足")
cleanup_disk_space()
else:
print(f"系统I/O错误: {e}")
finally:
logging.info("文件操作尝试完成") # 无论成功失败都会执行
三、高级异常处理技巧
1. 重试机制实现
import time
from functools import wraps
def retry_file_operation(max_retries=3, delay=1):
"""文件操作重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except (IOError, OSError) as e:
last_exception = e
if attempt < max_retries - 1:
time.sleep(delay * (attempt + 1))
continue
raise last_exception
return wrapper
return decorator
@retry_file_operation(max_retries=5, delay=0.5)
def safe_file_write(content, file_path):
"""带自动重试的文件写入"""
with open(file_path, 'w') as f:
f.write(content)
2. 上下文管理器进阶
class SafeFileOpener:
"""带完善异常处理的文件上下文管理器"""
def __init__(self, file_path, mode='r', encoding=None):
self.file_path = file_path
self.mode = mode
self.encoding = encoding
self.file = None
def __enter__(self):
try:
self.file = open(self.file_path, self.mode, encoding=self.encoding)
return self.file
except FileNotFoundError:
if 'r' in self.mode:
raise # 读取时文件必须存在
# 写入时尝试创建目录
os.makedirs(os.path.dirname(self.file_path), exist_ok=True)
self.file = open(self.file_path, self.mode, encoding=self.encoding)
return self.file
except PermissionError:
raise PermissionError(f"没有权限访问文件: {self.file_path}")
def __exit__(self, exc_type, exc_val, exc_tb):
if self.file:
self.file.close()
# 处理特定异常
if exc_type is UnicodeDecodeError:
raise ValueError("文件编码错误") from exc_val
return False # 不抑制其他异常
# 使用示例
try:
with SafeFileOpener('data/config.ini', 'r', encoding='utf-8') as f:
config = f.read()
except ValueError as e:
print(e)
3. 原子写入操作
import tempfile
import os
def atomic_write(file_path, content, encoding='utf-8'):
"""原子写入文件,避免写入过程中出错导致文件损坏"""
temp_fd, temp_path = tempfile.mkstemp(dir=os.path.dirname(file_path))
try:
with os.fdopen(temp_fd, 'w', encoding=encoding) as f:
f.write(content)
# 重命名是原子操作
os.replace(temp_path, file_path)
except Exception:
# 确保临时文件被清理
try:
os.unlink(temp_path)
except OSError:
pass
raise
四、特定场景的异常处理
1. 处理大文件时的异常
def process_large_file(file_path):
"""大文件处理中的异常处理"""
try:
file_size = os.path.getsize(file_path)
if file_size > 1_000_000_000: # >1GB
confirm = input("警告:处理大文件,确认继续?(y/n) ")
if confirm.lower() != 'y':
return
with open(file_path, 'rb') as f:
for chunk in iter(lambda: f.read(1024*1024), b''): # 每次1MB
try:
process(chunk)
except ProcessingError as e:
print(f"处理数据块时出错: {e}")
continue # 跳过错误块继续处理
except MemoryError:
print("内存不足,尝试使用更小的块处理")
# 回退策略
with open(file_path, 'rb') as f:
for chunk in iter(lambda: f.read(256*1024), b''): # 改为256KB
process(chunk)
2. 网络文件系统特殊处理
def handle_nfs_file(file_path):
"""处理网络文件系统(NFS)的特殊异常"""
max_retries = 3
for attempt in range(max_retries):
try:
with open(file_path, 'r+') as f:
# NFS可能出现的特殊错误
try:
data = f.read()
# 处理数据...
f.seek(0)
f.write(processed_data)
f.truncate()
break # 成功则退出循环
except OSError as e:
if e.errno == 121: # 远程I/O错误
time.sleep(1)
continue
raise
except FileNotFoundError:
if attempt == max_retries - 1:
raise
time.sleep(1)
3. 关键配置文件的容错处理
def load_critical_config(config_path):
"""关键配置文件的加载,带多重回退"""
config_locations = [
config_path,
f"/etc/{os.path.basename(config_path)}",
os.path.expanduser(f"~/.config/{os.path.basename(config_path)}")
]
for location in config_locations:
try:
with open(location, 'r', encoding='utf-8') as f:
try:
return json.load(f)
except json.JSONDecodeError:
# 尝试作为纯文本读取
f.seek(0)
return parse_alternative_config_format(f.read())
except (FileNotFoundError, PermissionError):
continue
# 所有位置都失败
raise RuntimeError("无法加载配置文件,所有尝试位置都失败")
五、异常处理最佳实践
- 精准捕获:只捕获你能处理的异常类型
# 不推荐
try:
file_op()
except: # 捕获所有异常,包括SystemExit
pass
# 推荐
try:
file_op()
except (IOError, OSError) as e: # 只捕获预期的I/O异常
handle_error(e)
- 异常上下文:使用raise from保留原始异常栈
try:
parse_config()
except ValueError as e:
raise ConfigError("Invalid config") from e
- 资源清理:确保文件句柄被释放
f = None
try:
f = open('file.txt')
# ...
finally:
if f is not None:
f.close()
- 错误日志:记录足够的调试信息
try:
save_data()
except Exception as e:
logging.error("保存数据失败: %s", e, exc_info=True)
logging.debug("失败时的系统状态: %s", get_system_status())
raise
- 用户友好消息:将技术异常转换为用户可理解的消息
error_messages = {
errno.ENOENT: "文件不存在",
errno.EACCES: "没有访问权限",
errno.ENOSPC: "磁盘空间不足"
}
try:
write_to_file()
except OSError as e:
print(error_messages.get(e.errno, f"系统错误: {e}"))
六、完整示例:安全的文件处理器
import os
import errno
import logging
from typing import Optional
class SafeFileHandler:
"""安全的文件操作处理器"""
def __init__(self, file_path: str):
self.file_path = file_path
self.backup_path = f"{file_path}.bak"
def read(self) -> Optional[str]:
"""安全读取文件内容"""
try:
with open(self.file_path, 'r', encoding='utf-8') as f:
return f.read()
except FileNotFoundError:
logging.warning("文件不存在: %s", self.file_path)
return None
except UnicodeDecodeError:
logging.error("文件编码错误: %s", self.file_path)
raise
except IOError as e:
logging.error("读取文件失败: %s [errno=%d]", e, e.errno)
raise
def write(self, content: str) -> bool:
"""安全写入文件,带备份和原子操作"""
try:
# 1. 备份原文件
if os.path.exists(self.file_path):
os.replace(self.file_path, self.backup_path)
# 2. 原子写入新文件
temp_fd, temp_path = tempfile.mkstemp(
dir=os.path.dirname(self.file_path),
prefix=os.path.basename(self.file_path))
try:
with os.fdopen(temp_fd, 'w', encoding='utf-8') as f:
f.write(content)
os.replace(temp_path, self.file_path)
return True
except Exception:
# 3. 恢复备份
if os.path.exists(self.backup_path):
os.replace(self.backup_path, self.file_path)
raise
finally:
# 确保临时文件被清理
if os.path.exists(temp_path):
try:
os.unlink(temp_path)
except OSError:
pass
except OSError as e:
logging.error("文件操作失败: %s [errno=%d]", e, e.errno)
if e.errno == errno.ENOSPC:
logging.critical("磁盘空间不足!")
return False
def __enter__(self):
"""上下文管理器支持"""
self.content = self.read()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""退出上下文时自动保存"""
if exc_type is None and hasattr(self, 'content'):
self.write(self.content)
return False
七、总结
- 始终对文件操作添加异常处理
- 区分不同类型的I/O错误并分别处理
- 确保资源释放,使用上下文管理器或finally块
- 考虑原子操作,避免文件损坏
- 提供有意义的错误信息和恢复方案
通过实现这些最佳实践,你的文件操作代码将更加健壮、可靠,能够应对各种异常情况
相关推荐
- sharding-jdbc实现`分库分表`与`读写分离`
-
一、前言本文将基于以下环境整合...
- 三分钟了解mysql中主键、外键、非空、唯一、默认约束是什么
-
在数据库中,数据表是数据库中最重要、最基本的操作对象,是数据存储的基本单位。数据表被定义为列的集合,数据在表中是按照行和列的格式来存储的。每一行代表一条唯一的记录,每一列代表记录中的一个域。...
- MySQL8行级锁_mysql如何加行级锁
-
MySQL8行级锁版本:8.0.34基本概念...
- mysql使用小技巧_mysql使用入门
-
1、MySQL中有许多很实用的函数,好好利用它们可以省去很多时间:group_concat()将取到的值用逗号连接,可以这么用:selectgroup_concat(distinctid)fr...
- MySQL/MariaDB中如何支持全部的Unicode?
-
永远不要在MySQL中使用utf8,并且始终使用utf8mb4。utf8mb4介绍MySQL/MariaDB中,utf8字符集并不是对Unicode的真正实现,即不是真正的UTF-8编码,因...
- 聊聊 MySQL Server 可执行注释,你懂了吗?
-
前言MySQLServer当前支持如下3种注释风格:...
- MySQL系列-源码编译安装(v5.7.34)
-
一、系统环境要求...
- MySQL的锁就锁住我啦!与腾讯大佬的技术交谈,是我小看它了
-
对酒当歌,人生几何!朝朝暮暮,唯有己脱。苦苦寻觅找工作之间,殊不知今日之事乃我心之痛,难道是我不配拥有工作嘛。自面试后他所谓的等待都过去一段时日,可惜在下京东上的小金库都要见低啦。每每想到不由心中一...
- MySQL字符问题_mysql中字符串的位置
-
中文写入乱码问题:我输入的中文编码是urf8的,建的库是urf8的,但是插入mysql总是乱码,一堆"???????????????????????"我用的是ibatis,终于找到原因了,我是这么解决...
- 深圳尚学堂:mysql基本sql语句大全(三)
-
数据开发-经典1.按姓氏笔画排序:Select*FromTableNameOrderByCustomerNameCollateChinese_PRC_Stroke_ci_as//从少...
- MySQL进行行级锁的?一会next-key锁,一会间隙锁,一会记录锁?
-
大家好,是不是很多人都对MySQL加行级锁的规则搞的迷迷糊糊,一会是next-key锁,一会是间隙锁,一会又是记录锁。坦白说,确实还挺复杂的,但是好在我找点了点规律,也知道如何如何用命令分析加...
- 一文讲清怎么利用Python Django实现Excel数据表的导入导出功能
-
摘要:Python作为一门简单易学且功能强大的编程语言,广受程序员、数据分析师和AI工程师的青睐。本文系统讲解了如何使用Python的Django框架结合openpyxl库实现Excel...
- 用DataX实现两个MySQL实例间的数据同步
-
DataXDataX使用Java实现。如果可以实现数据库实例之间准实时的...
- MySQL数据库知识_mysql数据库基础知识
-
MySQL是一种关系型数据库管理系统;那废话不多说,直接上自己以前学习整理文档:查看数据库命令:(1).查看存储过程状态:showprocedurestatus;(2).显示系统变量:show...
- 如何为MySQL中的JSON字段设置索引
-
背景MySQL在2015年中发布的5.7.8版本中首次引入了JSON数据类型。自此,它成了一种逃离严格列定义的方式,可以存储各种形状和大小的JSON文档,例如审计日志、配置信息、第三方数据包、用户自定...
你 发表评论:
欢迎- 一周热门
-
-
MySQL中这14个小玩意,让人眼前一亮!
-
旗舰机新标杆 OPPO Find X2系列正式发布 售价5499元起
-
【VueTorrent】一款吊炸天的qBittorrent主题,人人都可用
-
面试官:使用int类型做加减操作,是线程安全吗
-
C++编程知识:ToString()字符串转换你用正确了吗?
-
【Spring Boot】WebSocket 的 6 种集成方式
-
PyTorch 深度学习实战(26):多目标强化学习Multi-Objective RL
-
pytorch中的 scatter_()函数使用和详解
-
与 Java 17 相比,Java 21 究竟有多快?
-
基于TensorRT_LLM的大模型推理加速与OpenAI兼容服务优化
-
- 最近发表
- 标签列表
-
- idea eval reset (50)
- vue dispatch (70)
- update canceled (42)
- order by asc (53)
- spring gateway (67)
- 简单代码编程 贪吃蛇 (40)
- transforms.resize (33)
- redisson trylock (35)
- 卸载node (35)
- np.reshape (33)
- torch.arange (34)
- npm 源 (35)
- vue3 deep (35)
- win10 ssh (35)
- vue foreach (34)
- idea设置编码为utf8 (35)
- vue 数组添加元素 (34)
- std find (34)
- tablefield注解用途 (35)
- python str转json (34)
- java websocket客户端 (34)
- tensor.view (34)
- java jackson (34)
- vmware17pro最新密钥 (34)
- mysql单表最大数据量 (35)