本文主要是介绍file | 某文件夹【解耦合】下的文件查找功能实现及功能单元测试,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!
文件查找工具
- 概要
- 思路
- OS模块 --- 学习版
- os.getcwd()
- os.path.dirname(os.getcwd())
- os.path.dirname() 和 os.path.basename()
- OS模块 — 实战版
- 单元测试
- 解耦合
概要
- 梳理业务主逻辑:
查看存放被采集JSON数据的文件夹内的文件列表
【所有 包含文件夹下的文件夹下的文件】
这是本节内容聚焦的点- 和MySQL内记录的信息做比对,判断哪些文件是新文件,需要采集的。
- 读取JSON文件,执行ETL操作(读取->写入CSV->写入MySQL(目的地库))。
- 将被处理的JSON信息,记录到MySQL数据库(元数据库)
思路
读取某个文件中的文件名--------------------set()
读取数据库中已经处理过的文件名称-----set()
做处理 – 得到未处理过的文件名-----------set()-set()
如何处理?
- 思路1:集合相减得到的结果是在第一个集合且不在第二个集合中的元素
- 思路2: for循环
OS模块 — 学习版
- 耦合
和固定路径绑定在一起,若是将这部分代码给其他人,他们需要创建对应文件夹才可以使用
# 学习OS 接口模块
import os# 输出文件下的子文件 - 不包括子文件下的文件 仅返回子文件夹的名称
files = os.listdir(r'E:\pythonProject\ETL\day04_商品数据采集\04_数据\采集JSON')
print(files)# 输出文件下的子文件 包括子文件夹呢?
# 递归调用
def read_dir(dir):results =[]files = os.listdir(dir)for file in files:# 判断是否是目录if os.path.isdir(file):results += read_dir(dir+'/'+file)else:results.append(dir+'/'+file)return results
- 调用
print(read_dir(os.getcwd()))
['E:\\pythonProject\\pythonetl\\learning/learning_os.py', 'E:\\pythonProject\\pythonetl\\learning/learning_time.py', 'E:\\pythonProject\\pythonetl\\learning/learning_unittest.py', 'E:\\pythonProject\\pythonetl\\learning/learn_logging.py', 'E:\\pythonProject\\pythonetl\\learning/log_t1.py', 'E:\\pythonProject\\pythonetl\\learning/mycode.py']
当前路径E:\pythonProject\pythonetl\learning\learning_os.py,在learning_os.py中执行以下代码
os.getcwd()
print('getcwd',os.getcwd())
getcwd E:\pythonProject\pythonetl\learning
os.path.dirname(os.getcwd())
print('dirname',os.path.dirname(os.getcwd()))
dirname E:\pythonProject\pythonetl
os.path.dirname() 和 os.path.basename()
stra = ['E:\pythonProject\pythonetl']
for i in stra:print(os.path.dirname(i))print(os.path.basename(i))
E:\pythonProject
pythonetl
OS模块 — 实战版
- 在until文件夹下创建filr_until.py文件
import osdef get_dir_files_list(path="./",recursive=False):# 判断文件夹下面,有哪些文件# :param path:被判断的文件夹的路径,默认当前路径# :param recursive:是否递归读取,默认不递归# :return:1ist对象,list里面存储的是文件的路径#os.listdir这个API返回的是你给定的path下面有哪些`文件和文件夹`、dir_names = os.listdir(path)# print('dir_names',dir_names)# if not dir_names:# return os.path.basename(path)files =[]#定义一个list,用来记录文件for dir_name in dir_names:absolute_path = f"{path}/{dir_name}"if not os.path.isdir(absolute_path):# print('not a dir')#如果进来这个if,表明这个是:文件files.append(absolute_path)else:# print('is a dir')#表明是文件夹if recursive:#如果recursive是True,表明要进到文件夹里面继续找文件files += get_dir_files_list(absolute_path,recursive)return files
单元测试
- test文件夹下的test_file_until.py文件
import os.path
from unittest import TestCase
from util.file_util import get_dir_files_listclass TestFileUtil(TestCase):# 确定手动创建的测试目录的绝对路径def setUp(self) -> None: # 测试前需要提前执行的代码 例如连接数据库self.project_root_path = os.path.dirname(os.getcwd())print('解耦合获得路径:',self.project_root_path)passdef test_myfunc(self): # 要以test开头运行# 测试get_dir_files_list 函数# 该以哪一个作为我们的测试目录# 解耦合 ---'''请在工程根目录的test文件夹内建立:test_dir /inner1 /iner2 /innner3/53412的目录结构用于进行此方法的单元测试不递归结果应该是1和2递归结果应该是1, 2, 3, 4, 5'''# 测试没有开启递归调用的代码result1 = get_dir_files_list(path = self.project_root_path + '/' + 'test_dir',recursive=False)predicted_result= ['1', '2']# self.assertEqual(results, predicted_result)result_1 = []for p in predicted_result:result_1.append(self.project_root_path + '/' + 'test_dir'+'/'+p)# 排除顺序对结果的影响result1.sort()result_1.sort()# 这里断言 函数获得的结果和预期的结果路径是一致的self.assertEqual(result1, result_1)# 测试开启递归调用的代码 不包括inner3这种情况result2 = get_dir_files_list(path = self.project_root_path + '/' + 'test_dir',recursive=True)predicted_result = ['1', '2','inner1/3','inner1/4','inner1/inner2/5',]# 若是 'inner1/4','inner1/3' 则会出现错误 因此 我们要加上一道保险 调整顺序# 使用sort()函数 没有返回值result_2 = []for p in predicted_result:result_2.append(self.project_root_path + '/' + 'test_dir'+'/'+p)result2.sort()result_2.sort()self.assertEqual(result2, result_2)# 测试开启递归调用的代码 测试空文件夹 针对 inner3这种情况'''实际上输出 不应该出现 'E:\\pythonProject\\pythonetl/test_dir/inner1/inner2/inner3'这种情况因为我们遍历的是可以用的文件 而非目录 所以想办法排除目录'''result3 = get_dir_files_list(path = self.project_root_path + '/' + 'test_dir',recursive=True)predicted_result = ['1', '2','inner1/3','inner1/4','inner1/inner2/5','inner1/inner2/inner3'] ## 若是 'inner1/4','inner1/3' 则会出现错误 因此 我们要加上一道保险 调整顺序# 使用sort()函数 没有返回值result_3 = []for p in predicted_result:result_3.append(self.project_root_path + '/' + 'test_dir'+'/'+p)#if os.path.isdir(result_3[-1]):if not os.listdir(result_3[-1]):result_3.pop(-1)result3.sort()result_3.sort()self.assertEqual(result3, result_3)def tearDown(self)-> None: # 收尾工作pass
解耦合
在软件开发中,解耦合(Decoupling)是指减少或去除系统中各组件之间的相互依赖关系,以提高系统的灵活性和可维护性。
解耦合的目标是创建松散耦合的系统,其中每个组件或模块都可以独立于其他组件进行开发、测试和维护。
例如:
我的一个系统的文件路径是绝对路径, 若是我将该系统给另一个人,他会因为文件夹是否存在的问题而出现报错的风险,因此我们要做解耦合的措施
这篇关于file | 某文件夹【解耦合】下的文件查找功能实现及功能单元测试的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!