智能创建多级目录
from boltons.fileutils import mkdir_p
# 一次性创建多级目录
mkdir_p("/home/user/projects/new_project/src")
原子文件保存
from boltons.fileutils import atomic_save
with atomic_save('important_data.json') as f:
f.write(json.dumps(data))
即使在写入过程中发生系统崩溃或断电,也能保证文件不会损坏,要么保存成功,要么保留原文件。
高级迭代处理
Boltons的iterutils模块提供了强大的迭代工具,大大简化了复杂数据处理
from boltons.iterutils import remap, research
# 删除嵌套结构中的所有None值
data ={'a':{'b':1,'c':None},'d':None}
cleaned_data = remap(data,lambda p, k, v: v isnotNone)
print(cleaned_data)# 输出:{'a': {'b': 1}}
# 递归查找所有整数值
nested_dict ={'a':{'b':1,'c':(2,'x',3)},'d':None}
int_values = research(nested_dict, query=lambda p, k, v:isinstance(v,int))
print(list(int_values)) # 输出:[(('a','b'), 1), (('a','c',0), 2), (('a','c',2), 3)]
remap函数特别适合处理JSON等嵌套数据结构,而research则提供了强大的递归搜索能力。
高效分块处理大数据
当处理大型数据集时,分块处理是提高效率的关键,这种方法比一次性处理整个数据集内存占用减少90%以上,特别适合处理大型文件或数据库查询结果。
from boltons.iterutils import chunked
big_data = list(range(1000))
# 将大数据集分成每批100个
for batch in chunked(big_data, 100):
process_batch(batch)
智能日期范围生成
通过调整step参数((年,月,日)元组),可以灵活控制日期步长。
from datetime import date
from boltons.timeutils import daterange
start_date = date(2023,1,1)
end_date = date(2023,1,10)
# 生成日期范围
for day in daterange(start_date, end_date):
print(day)
# 按月生成
for month_day in daterange(date(2023,1,1), date(2023,4,1),
step=(0,1,0)):
print(month_day) # 输出每个月的第一天
字符串智能解析
大幅减少数据清洗和预处理的时间
from boltons.strutils import parse_int_list, camel2under
# 解析复杂数字范围
cpu_cores = parse_int_list('0,7,21-22,48,55,69-70')
print(cpu_cores) # 输出:[0, 7, 21, 22, 48, 55, 69, 70]
# 驼峰命名转下划线命名
class_name = 'MyAwesomeClass'
print(camel2under(class_name)) # 输出:'my_awesome_class'
有序集合(IndexedSet)
Python标准库的set不保留元素顺序,这在某些场景下很不方便。Boltons提供了有序集合
from boltons.setutils importIndexedSet
# 创建有序集合并保留元素添加顺序
data =IndexedSet(list(range(3,6))+list(range(2)))
print(data)# 输出:IndexedSet([3, 4, 5, 0, 1])
# 支持集合操作同时保持顺序
set1 =IndexedSet([1,3,5])
set2 =IndexedSet([2,3,4])
print(set1 | set2) # 输出:IndexedSet([1, 3, 5, 2, 4])
高级缓存管理
Boltons的cacheutils模块提供了多种缓存策略,对于计算成本高的操作,合理使用缓存可以提升程序性能10倍以上。
from boltons.cacheutils import LRU
# 创建最近最少使用缓存
cache = LRU(max_size=1000)
# 添加缓存项
cache['query_result'] = expensive_query()
# 获取缓存
if 'query_result' in cache:
result = cache['query_result']
结构化URL解析
from boltons.urlutils import URL
url = URL('https://user:[email protected]:8042/path?query=param#fragment')
print(url.scheme) # 输出:'https'
print(url.host) # 输出:'example.com'
print(url.port) # 输出:8042
print(url.path) # 输出:'/path'
print(url.query) # 输出:'query=param'
print(url.fragment) # 输出:'fragment'
多文件流式读取
这在处理分块上传的文件或日志文件拼接时特别有用。
from boltons.ioutils import MultiFileReader
from io import BytesIO
file1 = BytesIO(b'First file content')
file2 = BytesIO(b'Second file content')
mfr = MultiFileReader(file1, file2)
combined_content = mfr.read().decode('utf-8')
高级属性缓存
from boltons.cacheutils import cachedproperty
class DataProcessor:
def__init__(self, data):
self.data = data
@cachedproperty
def processed_data(self):
print("Processing data...")
# 复杂的数据处理过程
returnsorted(self.data)
processor =DataProcessor([5,2,8,1])
print(processor.processed_data)# 第一次调用会计算
print(processor.processed_data) # 第二次直接返回缓存结果
增强的异常信息
from boltons.tbutils importTracebackInfo
try:
# 可能出错的代码
1/0
exceptExceptionas e:
tb_info =TracebackInfo.from_exception(e)
print(f"Exception occurred at {tb_info.frames[-1].filename}:{tb_info.frames[-1].lineno}")