我在使用函数式编程时，遇到了一个问题，背景是这样的：一个项目中存在了大量的函数，这些函数之间的相互调用，使得导致代码逻辑比较复杂。当我想要修改一个逻辑的时候，需要反复的查看代码，以及定义新的函数。
函数相对独立，本身是一个很好的特点。然而，正是因为相互独立，使得函数和函数的依赖关系缺失，反而造成了程序设计与改动的不方便。
当我们使用函数的时候，为了能够较好的修改原本的逻辑，可以通过默认参数来构成依赖关系。参数的可替代性，可以使得代码更容易被修改。 但是相比于 OOP 机制而言，FP 的依赖关系不那么显然易见——基于类的成员函数代码，让工程师更容易理解依赖关系。并且，在函数跳转的时候，借助ide或者编辑器，跳转也更加方便。
然而我这个想法真的正确吗？是否是因为函数编程本身有一些特性我没有使用到呢？
代码
我遇到问题，代码如下：
函数式代码
@validate
def compute_single(
get_id_and_params: t.Callable[[], tuple[int, T.general.Params]],
get_full_path: t.Callable[[str], str],
*,
func_set: BasicFuncSet = FuncSet(),
is_write: bool = True,
) -> tuple[result.ComputeResult, pd.DataFrame]:
"""compute single values.
is_write: write to file
"""
id, params = get_id_and_params()
input_data = InputData.from_dict(params)
filelist: t.List[result.ResultFile] = []
# 生成写入数据的函数
write_data = use_write_data(
id,
params["udl2_file"],
filelist.append,
current_type=1 if is_zhiliu(params) else 0, # ac 0, dc 1. zhiliu is dc
)
# filter period
df = func_set.filter(input_data=input_data)
if is_write:
write_data(df, params["udl2_file"], get_full_path(params["out_file_path_1"]))
# stat period, support multiple files, but not use this is api call
df = (func_set.stat(df, params, interval_jihua=params.get("interval_jihua", True)),)
df = (dict(iter(row)) for row in df)
df = pd.DataFrame(df)
if is_write:
write_data(
df, params["out_file_path_1"], get_full_path(params["out_file_path_2"])
)
# model period
df = func_set.model(df, params)
if is_write:
write_data(
df, params["out_file_path_2"], get_full_path(params["out_file_path_3"])
)
# 包含了三级文件
file_result = result.ComputeResult(
status=result.Status.success,
msg="success!",
data=filelist,
)
# 返回文件以及最终的DataFrame
return file_result, df
def write_final_files(
ac_df: pd.DataFrame,
dc_df: pd.DataFrame,
output_path_fn: t.Callable[[str], str],
) -> result.FinalFiles:
"""写入 ac dc 文件"""
ac_file = output_path_fn("ac")
dc_file = output_path_fn("dc")
fileutils.to_csv(ac_df, str(ac_file))
fileutils.to_csv(dc_df, str(dc_file))
return result.FinalFiles(
ac=result.ResultFile(
id=1,
current_type=0,
original_filename="all udl2_files",
original_filepath="null",
in_filename="all results of udl2_files",
filename=fileutils.get_file_name(str(ac_file)),
filepath=str(ac_file),
),
dc=result.ResultFile(
id=2,
current_type=1,
original_filename="all udl2_files",
original_filepath="null",
in_filename="all results of udl2_files",
filename=fileutils.get_file_name(str(dc_file)),
filepath=str(dc_file),
),
)
def collect_files(
param_group: T.general.ParamGroup,
is_write: bool = True,
compute_single: T.ComputeFunc = compute_single,
) -> tuple[list[result.ComputeResult], list[pd.DataFrame]]:
"""
对参数列表里面的数据进行计算，并返回结果列表
compute_single: 计算单一文件的函数
"""
res: t.List[result.ComputeResult] = []
df_list: t.List[pd.DataFrame] = []
for i, p in enumerate(param_group["param_list"]):
def get_id_and_params():
return i, p
r, df = compute_single(
get_id_and_params,
param_group["get_full_path"],
is_write=is_write,
)
# r contains stage-1, stage-2, stage-3 files
res.append(r)
df_list.append(df)
return res, df_list
def compute(
param_group: T.general.ParamGroup,
is_write: bool = True,
collect_files: T.CollectFilesFunc = collect_files,
) -> result.FinalResult:
"""
计算并返回最终结果
collect_files 是收集文件的函数，也包含了计算单个文件的过程
"""
res, df_list = collect_files(param_group, is_write)
# 整理最终的文件
ac_df, dc_df = fileutils.merge_files(res, df_list)
output_fn = param_group.get("output_path")
if output_fn is None:
raise ValueError("output_path is None")
final_files = write_final_files(ac_df, dc_df, output_fn)
return result.FinalResult(
status=result.Status.success,
msg="success!",
compute_result=res,
final_files=final_files,
)
当我需要调整 funcset 的实现时，如果我想对当前的代码进行复用，会写成这样
def get_compute(funcset: BasicFuncSet):
"""隐藏了其他函数的复杂性，直接调用本函数即可完成 compute 函数的生成"""
return functools.partial(
controller.compute,
collect_files=functools.partial(
controller.collect_files,
compute_single=functools.partial(
controller.compute_single,
func_set=funcset,
),
),
)
为了能够修改funset中的内容，我需要不停的重新查看代码，需要不停的查文档，然后来编写代码。尽管这里使用了柯里化来简化代码的参数，但在我看来，实际上帮助不是很大。我仍然需要不停的查询函数的参数，从而降低了编程的速度。
基于 OOP（面向对象编程）的代码
class Controller(object):
def __init__(self, funcset: BasicFuncSet = FuncSet()):
self.funcset = funcset
@validate
def compute_single(
self,
get_id_and_params: t.Callable[[], tuple[int, T.general.Params]],
get_full_path: t.Callable[[str], str],
*,
is_write: bool = True,
) -> tuple[result.ComputeResult, pd.DataFrame]:
"""compute single values.
is_write: write to file
"""
id, params = get_id_and_params()
input_data = InputData.from_dict(params)
filelist: t.List[result.ResultFile] = []
# 生成写入数据的函数
write_data = use_write_data(
id,
params["udl2_file"],
filelist.append,
current_type=1 if is_zhiliu(params) else 0, # ac 0, dc 1. zhiliu is dc
)
# filter period
df = self.funcset.filter(input_data=input_data)
if is_write:
write_data(
df, params["udl2_file"], get_full_path(params["out_file_path_1"])
)
# stat period, support multiple files, but not use this is api call
df = (
self.funcset.stat(
df, params, interval_jihua=params.get("interval_jihua", True)
),
)
df = (dict(iter(row)) for row in df)
df = pd.DataFrame(df)
if is_write:
write_data(
df, params["out_file_path_1"], get_full_path(params["out_file_path_2"])
)
# model period
df = self.funcset.model(df, params)
if is_write:
write_data(
df, params["out_file_path_2"], get_full_path(params["out_file_path_3"])
)
# 包含了三级文件
file_result = result.ComputeResult(
status=result.Status.success,
msg="success!",
data=filelist,
)
# 返回文件以及最终的DataFrame
return file_result, df
def write_final_files(
self,
ac_df: pd.DataFrame,
dc_df: pd.DataFrame,
output_path_fn: t.Callable[[str], str],
) -> result.FinalFiles:
"""写入 ac dc 文件"""
ac_file = output_path_fn("ac")
dc_file = output_path_fn("dc")
fileutils.to_csv(ac_df, str(ac_file))
fileutils.to_csv(dc_df, str(dc_file))
return result.FinalFiles(
ac=result.ResultFile(
id=1,
current_type=0,
original_filename="all udl2_files",
original_filepath="null",
in_filename="all results of udl2_files",
filename=fileutils.get_file_name(str(ac_file)),
filepath=str(ac_file),
),
dc=result.ResultFile(
id=2,
current_type=1,
original_filename="all udl2_files",
original_filepath="null",
in_filename="all results of udl2_files",
filename=fileutils.get_file_name(str(dc_file)),
filepath=str(dc_file),
),
)
def collect_files(
self,
param_group: T.general.ParamGroup,
is_write: bool = True,
) -> tuple[list[result.ComputeResult], list[pd.DataFrame]]:
"""
对参数列表里面的数据进行计算，并返回结果列表
compute_single: 计算单一文件的函数
"""
res: t.List[result.ComputeResult] = []
df_list: t.List[pd.DataFrame] = []
for i, p in enumerate(param_group["param_list"]):
def get_id_and_params():
return i, p
r, df = self.compute_single(
get_id_and_params,
param_group["get_full_path"],
is_write=is_write,
)
# r contains stage-1, stage-2, stage-3 files
res.append(r)
df_list.append(df)
return res, df_list
def compute(
self,
param_group: T.general.ParamGroup,
is_write: bool = True,
) -> result.FinalResult:
"""
计算并返回最终结果
collect_files 是收集文件的函数，也包含了计算单个文件的过程
"""
res, df_list = self.collect_files(param_group, is_write)
# 整理最终的文件
ac_df, dc_df = fileutils.merge_files(res, df_list)
output_fn = param_group.get("output_path")
if output_fn is None:
raise ValueError("output_path is None")
final_files = self.write_final_files(ac_df, dc_df, output_fn)
return result.FinalResult(
status=result.Status.success,
msg="success!",
compute_result=res,
final_files=final_files,
)
相比之下，基于类的编程，我只需要这样做：
def get_compute(funcset: BasicFuncSet):
"""隐藏了其他函数的复杂性，直接调用本函数即可完成 compute 函数的生成"""
c = controller.Controller(funcset)
return c.compute
进一步分析
实际上，为了实现新需求，修改计算逻辑，对 funcset 的修改是不可或缺的。然而，第一种函数式编程的设计，并非没有显示的表现出函数之间的依赖关系，而是使得依赖被隐藏起来了。也就是说，关键的逻辑，被放在了调用栈深处。
那么，能否将funcset放在可以被简单修改的地方呢？
原来的代码是什么样子的？
get_compute这个方法不存在
可以采用 Controller 的方式。只不过修改代码的时候，需要查看每个方法的类型。
区别是，我们愿不愿意根据原本的调用链条，来重新组织代码。使用编辑器 DEBUG 模式，就能看到完整的调用链条。
在当前阶段明确问题，然后寻找解决办法。并非一股劲儿盲干。
依然函数式
实际上，以下这段代码就起到了这个作用。
def get_compute(funcset: BasicFuncSet):
"""隐藏了其他函数的复杂性，直接调用本函数即可完成 compute 函数的生成"""
return functools.partial(
controller.compute,
collect_files=functools.partial(
controller.collect_files,
compute_single=functools.partial(
controller.compute_single,
func_set=funcset,
),
),
)
实际上，尽管看起来代码量很多，就做了一件事情。把 compute_single 中的参数改成了 funcset。
还有一种方法，就是将默认参数拿出来。
def collect_files(
param_group: T.general.ParamGroup,
is_write: bool = True,
compute_single: T.ComputeFunc = compute_single,
funcset: BasicFuncSet = FuncSet(),
) -> tuple[list[result.ComputeResult], list[pd.DataFrame]]:
# ignore
使用 OOP（面向对象编程）
“相比之下，如果直接使用
Controller聚合为一个类，还是更加清晰一些。因此，此处使用面向类编程的设计，更好的适应项目不停修改的需求。” ——这是我之前的想法。
使用 OOP 并没有更加清晰。只不过是将原本明确的，函数之间的依赖关系隐藏起来了。这反而是没有必要的。当工程师看到这段代码的时候，只能看到一个Controller被初始化。实际上意义不大。
这样可能使得代码失去了函数式编程的灵活性。
Besides …
然而，如果我一开始就使用面向对象的设计，funcset 也不见得一定暴露在外（能够被轻松的修改）。可能也会引入其他的复杂性。如果使用类，则应该更好的去定义子类，让代码变得更加可读。除此之外，还有其他的复杂性。例如，使用类的方式，会使得测试代码的编写变得相对困难，需要引入 mock 等方法。如果想要简化，可能还得利用结构化的数据进行简化，例如 from pydantic import BaseModel，去生成yaml文件。
第一次编写代码的时候…
我在最初编写这段代码的时候能没有想到这段代码需要被复用。是在后续的逻辑中，对代码的逻辑进行调整，从而使得其能够被复用。换句话说，不论采用何种编程模式，测试+重构是让代码质量提升，更快的修改代码的一个很好的途径。一个很明显的例子是，sqlite 虽然是开源的，但是其测试套件是专有的。
其他的问题
- 当
get_compute方法被写好之后，Controller这个改动还是有必要的吗？
可能不是必要的。如果需要进一步了解相关的代码，一定会阅读 get_compute 的代码。相比之下，Controller反而更加复杂。因为 get_compute 已经将代码的调用关系非常清晰的展示出来了。
如果要修改，有两个思路：
- 如果是OOP，使用工具，查看
funcset在Controller中被引用的地方。 - 如果是FP，直接查看
compute_single函数。
第一时间想看简单的东西，还是直接看逻辑。我认为 FP 的方式更加友好，可控性更强。
结论
还是使用 FP 继续开展后续的工作。