通用工具类与异常处理:Python实用指南
- 1. 通用异常类:`ValidationException`
- 2. 通用工具类:`Utils`
- 3. 示例文件内容
- `default_config.yaml`
- `config.yaml`
- `data.jsonl`
- 4. 代码示例
- 5. 运行结果
- 6. 完整代码
- 7. 总结
在Python开发中,编写可重用的工具类和通用的异常处理机制是提高代码质量和开发效率的关键。本文将介绍如何将特定的异常类GlueValidaionException
改写为更通用的ValidationException
,并创建一个通用的工具类Utils
,包含常用的文件操作和数据处理方法。最后,我们将通过代码示例展示如何使用这些工具。
1. 通用异常类:ValidationException
首先,我们将GlueValidaionException
改写为更通用的ValidationException
,使其适用于各种验证场景。
class ValidationException(Exception):
"""
通用验证异常类,适用于各种验证场景。
"""
def __init__(self, err_message, excep_obj):
message = ("[Invalid input detected]\n"
f"Exception: {err_message}\n"
f"Exception logs: {excep_obj}")
super().__init__(message)
2. 通用工具类:Utils
接下来,我们创建一个通用的工具类Utils
,包含常用的文件操作和数据处理方法。
import json
from os.path import join
from typing import Dict, List
import yaml
class Utils:
@staticmethod
def yaml_to_dict(file_path: str) -> Dict:
"""
将YAML文件转换为字典。
:param file_path: YAML文件路径
:return: 解析后的字典
"""
with open(file_path) as yaml_file:
yaml_string = yaml_file.read()
try:
parsed_dict = yaml.safe_load(yaml_string)
except yaml.scanner.ScannerError as e:
raise ValidationException(f"YAML文件语法错误: {file_path}", e)
return parsed_dict
@staticmethod
def yaml_to_class(yaml_file_path: str, cls: type, default_yaml_file_path: str = None):
"""
将YAML文件转换为类对象。
:param yaml_file_path: YAML文件路径
:param cls: 目标类
:param default_yaml_file_path: 默认YAML文件路径
:return: 类对象
"""
if not yaml_file_path:
yaml_file_path = default_yaml_file_path
custom_args = Utils.yaml_to_dict(yaml_file_path)
if default_yaml_file_path:
default_args = Utils.yaml_to_dict(default_yaml_file_path)
missing_args = set(default_args) - set(custom_args)
for key in list(missing_args):
custom_args[key] = default_args[key]
try:
yaml_as_class = cls(**custom_args)
except TypeError as e:
raise ValidationException(f"YAML文件转换为类失败: {yaml_file_path}", e)
return yaml_as_class
@staticmethod
def read_jsonl(file_path: str) -> List:
"""
读取JSONL文件并返回所有JSON对象列表。
:param file_path: JSONL文件路径
:return: JSON对象列表
"""
jsonl_list = []
with open(file_path, "r") as fileobj:
while True:
single_row = fileobj.readline()
if not single_row:
break
json_object = json.loads(single_row.strip())
jsonl_list.append(json_object)
return jsonl_list
@staticmethod
def read_jsonl_row(file_path: str):
"""
逐行读取JSONL文件并返回JSON对象。
:param file_path: JSONL文件路径
:return: JSON对象生成器
"""
with open(file_path, "r") as fileobj:
while True:
try:
single_row = fileobj.readline()
if not single_row:
break
json_object = json.loads(single_row.strip())
yield json_object
except json.JSONDecodeError as e:
print(f"JSONL文件读取错误: {file_path}. 错误: {e}")
continue
@staticmethod
def append_as_jsonl(file_path: str, args_to_log: Dict):
"""
将字典追加到JSONL文件中。
:param file_path: JSONL文件路径
:param args_to_log: 要追加的字典
"""
json_str = json.dumps(args_to_log, default=str)
with open(file_path, "a") as fileobj:
fileobj.write(json_str + "\n")
@staticmethod
def save_jsonlist(file_path: str, json_list: List, mode: str = "a"):
"""
将JSON对象列表保存到JSONL文件中。
:param file_path: JSONL文件路径
:param json_list: JSON对象列表
:param mode: 文件写入模式
"""
with open(file_path, mode) as file_obj:
for json_obj in json_list:
json_str = json.dumps(json_obj, default=str)
file_obj.write(json_str + "\n")
@staticmethod
def str_list_to_dir_path(str_list: List[str]) -> str:
"""
将字符串列表拼接为目录路径。
:param str_list: 字符串列表
:return: 拼接后的目录路径
"""
if not str_list:
return ""
path = ""
for dir_name in str_list:
path = join(path, dir_name)
return path
3. 示例文件内容
以下是示例文件的内容,包括default_config.yaml
、config.yaml
和data.jsonl
。
default_config.yaml
name: "default_name"
value: 100
description: "This is a default configuration."
config.yaml
name: "custom_name"
value: 200
data.jsonl
{"id": 1, "name": "Alice", "age": 25}
{"id": 2, "name": "Bob", "age": 30}
{"id": 3, "name": "Charlie", "age": 35}
4. 代码示例
以下是如何使用Utils
工具类的示例:
# 示例类
class Config:
def __init__(self, name, value, description=None):
self.name = name
self.value = value
self.description = description
# 示例1: 将YAML文件转换为字典
yaml_dict = Utils.yaml_to_dict("config.yaml")
print("YAML文件转换为字典:", yaml_dict)
# 示例2: 将YAML文件转换为类对象
config_obj = Utils.yaml_to_class("config.yaml", Config, "default_config.yaml")
print("YAML文件转换为类对象:", config_obj.name, config_obj.value, config_obj.description)
# 示例3: 读取JSONL文件
jsonl_list = Utils.read_jsonl("data.jsonl")
print("读取JSONL文件:", jsonl_list)
# 示例4: 逐行读取JSONL文件
print("逐行读取JSONL文件:")
for json_obj in Utils.read_jsonl_row("data.jsonl"):
print(json_obj)
# 示例5: 将字典追加到JSONL文件
Utils.append_as_jsonl("data.jsonl", {"id": 4, "name": "David", "age": 40})
print("追加数据到JSONL文件完成")
# 示例6: 将JSON对象列表保存到JSONL文件
Utils.save_jsonlist("data.jsonl", [{"id": 5, "name": "Eve", "age": 45}])
print("保存JSON列表到JSONL文件完成")
# 示例7: 将字符串列表拼接为目录路径
path = Utils.str_list_to_dir_path(["dir1", "dir2", "dir3"])
print("拼接目录路径:", path)
5. 运行结果
运行上述代码后,输出结果如下:
YAML文件转换为字典: {'name': 'custom_name', 'value': 200}
YAML文件转换为类对象: custom_name 200 This is a default configuration.
读取JSONL文件: [{'id': 1, 'name': 'Alice', 'age': 25}, {'id': 2, 'name': 'Bob', 'age': 30}, {'id': 3, 'name': 'Charlie', 'age': 35}]
逐行读取JSONL文件:
{'id': 1, 'name': 'Alice', 'age': 25}
{'id': 2, 'name': 'Bob', 'age': 30}
{'id': 3, 'name': 'Charlie', 'age': 35}
追加数据到JSONL文件完成
保存JSON列表到JSONL文件完成
拼接目录路径: dir1/dir2/dir3
6. 完整代码
import json
from os.path import join
from typing import Dict, List
import yaml
from yaml.scanner import ScannerError
class ValidationException(Exception):
"""
通用验证异常类,适用于各种验证场景。
"""
def __init__(self, err_message, excep_obj):
message = ("[Invalid input detected]\n"
f"Exception: {err_message}\n"
f"Exception logs: {excep_obj}")
super().__init__(message)
class Utils:
@staticmethod
def yaml_to_dict(file_path: str) -> Dict:
"""
将YAML文件转换为字典。
:param file_path: YAML文件路径
:return: 解析后的字典
"""
with open(file_path) as yaml_file:
yaml_string = yaml_file.read()
try:
parsed_dict = yaml.safe_load(yaml_string)
except ScannerError as e:
raise ValidationException(f"YAML文件语法错误: {file_path}", e)
return parsed_dict
@staticmethod
def yaml_to_class(yaml_file_path: str, cls: type, default_yaml_file_path: str = None):
"""
将YAML文件转换为类对象。
:param yaml_file_path: YAML文件路径
:param cls: 目标类
:param default_yaml_file_path: 默认YAML文件路径
:return: 类对象
"""
if not yaml_file_path:
yaml_file_path = default_yaml_file_path
custom_args = Utils.yaml_to_dict(yaml_file_path)
if default_yaml_file_path:
default_args = Utils.yaml_to_dict(default_yaml_file_path)
missing_args = set(default_args) - set(custom_args)
for key in list(missing_args):
custom_args[key] = default_args[key]
try:
yaml_as_class = cls(**custom_args)
except TypeError as e:
raise ValidationException(f"YAML文件转换为类失败: {yaml_file_path}", e)
return yaml_as_class
@staticmethod
def read_jsonl(file_path: str) -> List:
"""
读取JSONL文件并返回所有JSON对象列表。
:param file_path: JSONL文件路径
:return: JSON对象列表
"""
jsonl_list = []
with open(file_path, "r") as file_obj:
while True:
single_row = file_obj.readline()
if not single_row:
break
json_obj = json.loads(single_row.strip())
jsonl_list.append(json_obj)
return jsonl_list
@staticmethod
def read_jsonl_row(file_path: str):
"""
逐行读取JSONL文件并返回JSON对象。
:param file_path: JSONL文件路径
:return: JSON对象生成器
"""
with open(file_path, "r") as file_object:
while True:
try:
single_row = file_object.readline()
if not single_row:
break
json_obj = json.loads(single_row.strip())
yield json_obj
except json.JSONDecodeError as e:
print(f"JSONL文件读取错误: {file_path}. 错误: {e}")
continue
@staticmethod
def append_as_jsonl(file_path: str, args_to_log: Dict):
"""
将字典追加到JSONL文件中。
:param file_path: JSONL文件路径
:param args_to_log: 要追加的字典
"""
json_str = json.dumps(args_to_log, default=str)
with open(file_path, "a") as file_object:
file_object.write(json_str + "\n")
@staticmethod
def save_jsonlist(file_path: str, json_list: List, mode: str = "a"):
"""
将JSON对象列表保存到JSONL文件中。
:param file_path: JSONL文件路径
:param json_list: JSON对象列表
:param mode: 文件写入模式
"""
with open(file_path, mode) as file_obj:
for json_obj in json_list:
json_str = json.dumps(json_obj, default=str)
file_obj.write(json_str + "\n")
@staticmethod
def str_list_to_dir_path(str_list: List[str]) -> str:
"""
将字符串列表拼接为目录路径。
:param str_list: 字符串列表
:return: 拼接后的目录路径
"""
if not str_list:
return ""
dir_path = ""
for dir_name in str_list:
dir_path = join(dir_path, dir_name)
return dir_path
# 示例类
class Config:
def __init__(self, name, value, description=None):
self.name = name
self.value = value
self.description = description
# 示例1: 将YAML文件转换为字典
yaml_dict = Utils.yaml_to_dict("config.yaml")
print("YAML文件转换为字典:", yaml_dict)
# 示例2: 将YAML文件转换为类对象
config_obj = Utils.yaml_to_class("config.yaml", Config, "default_config.yaml")
print("YAML文件转换为类对象:", config_obj.name, config_obj.value, config_obj.description)
# 示例3: 读取JSONL文件
jsonl_list = Utils.read_jsonl("data.jsonl")
print("读取JSONL文件:", jsonl_list)
# 示例4: 逐行读取JSONL文件
print("逐行读取JSONL文件:")
for json_obj in Utils.read_jsonl_row("data.jsonl"):
print(json_obj)
# 示例5: 将字典追加到JSONL文件
Utils.append_as_jsonl("data.jsonl", {"id": 4, "name": "David", "age": 40})
print("追加数据到JSONL文件完成")
# 示例6: 将JSON对象列表保存到JSONL文件
Utils.save_jsonlist("data.jsonl", [{"id": 5, "name": "Eve", "age": 45}])
print("保存JSON列表到JSONL文件完成")
# 示例7: 将字符串列表拼接为目录路径
path = Utils.str_list_to_dir_path(["dir1", "dir2", "dir3"])
print("拼接目录路径:", path)
7. 总结
通过将特定的异常类改写为通用的ValidationException
,并创建一个包含常用方法的Utils
工具类,我们可以大大提高代码的复用性和可维护性。本文提供的代码示例展示了如何使用这些工具类进行文件操作和数据处理,适合初级Python程序员学习和参考。